Repository: sktime/sktime
Pull request #6453 (2024-05-20 05:46:40)

[ENH] Apply forecasters depending on series type (e.g., intermittent) compositor
#### Reference Issues/PRs
<!--Implements a compositor which accepts a series-to-primitives transformer as a parameter; fixes #6279 -->

#### What does this implement/fix? Explain your changes.
This implementation introduces the `TransformSelectForecaster` which accepts a series-to-primitives transformer and a dictionary of forecasters as initialization parameters. This feature will allow users to categorize their input data through a transformer and then apply specific forecasters best suited for the classification categories.

#### What should a reviewer concentrate their feedback on?

A reviewer should most focus on the `_category_compositor.py` file in the `sktime/forecasting/compose` directory! It might also be helpful to take a look at the `ADICVTransformer` as it is a default parameter for this compositor.

#### Did you add any tests for the change?

I have not implemented any tests yet, but I am planning on adding them once I can get more feedback about the implementation and some advice on good test cases for this particular function!

#### PR checklist


##### For all contributions
- [x] I've added myself to the [list of contributors](https://github.com/sktime/sktime/blob/main/CONTRIBUTORS.md) with any new badges I've earned :-)
  How to: add yourself to the [all-contributors file](https://github.com/sktime/sktime/blob/main/.all-contributorsrc) in the `sktime` root directory (not the `CONTRIBUTORS.md`). Common badges: `code` - fixing a bug, or adding code logic. `doc` - writing or improving documentation or docstrings. `bug` - reporting or diagnosing a bug (get this plus `code` if you also fixed the bug in the PR).`maintenance` - CI, test framework, release.
  See here for [full badge reference](https://allcontributors.org/docs/en/emoji-key)
- [x] Optionally, for added estimators: I've added myself and possibly to the `maintainers` tag - do this if you want to become the owner or maintainer of an estimator you added.
  See here for further details on the [algorithm maintainer role](https://www.sktime.net/en/latest/get_involved/governance.html#algorithm-maintainers).
- [x] The PR title starts with either [ENH], [MNT], [DOC], or [BUG]. [BUG] - bugfix, [MNT] - CI, test framework, [ENH] - adding or improving code, [DOC] - writing or improving documentation or docstrings.

##### For new estimators
- [x] I've added the estimator to the API reference - in `docs/source/api_reference/taskname.rst`, follow the pattern.
- [x] I've added one or more illustrative usage examples to the docstring, in a pydocstyle compliant `Examples` section.
- [x] If the estimator relies on a soft dependency, I've set the `python_dependencies` tag and ensured
  dependency isolation, see the [estimator dependencies guide](https://www.sktime.net/en/latest/developer_guide/dependencies.html#adding-a-soft-dependency).

shlok191: @fkiraly, Hello Franz, I've made the changes that we discussed prior! I have made the series-to-primitives transformer a passable parameter and attempted to generalize the model. I had two questions I wanted to discuss though:

- For the passed transformer, is it safe for me to assume that the first column from a `transform` function will always include the data I require to choose a forecaster?

- For the test cases, would you have an idea about what might be good scenarios? I can think about instances that might lead to clear failures (like not passing in all required forecasters) but I think there can be better ones! 

Thanks a lot for the help!
shlok191: @fkiraly, I am running the pytests locally and did not encounter the issues I'm seeing online. Could you please tell me if I'm doing something wrong with the soft dependencies? Thanks!
fkiraly: > I am running the pytests locally and did not encounter the issues I'm seeing online. Could you please tell me if I'm doing something wrong with the soft dependencies? Thanks!

Which failures precisely, can you describe?
fkiraly: Re failures currently, I would suggest you use `check_estimator` to debug them, see guide here:
https://www.sktime.net/en/latest/developer_guide/add_estimators.html#using-the-check-estimator-utility

Feel free to ask in the `dev-chat` discord channel if you get stuck
shlok191: @benHeid, @astrogilda Hey guys! I'm nearly done working on this PR.

I've received some great feedback from Franz which I've tried to incorporate, and I'll really appreciate it if y'all could take a look at my code as well. 

Thank you so much!
fkiraly: Sure! short remark, the "protocol" for pull request review is that you should comment on the open discussions, one of:

* explain how addressed
* argue that needs not be addressed
* request information or discussion

Until consensus is reached.
shlok191: @fkiraly, instead of using `HierarchyEnsembleForecaster` for which I am a little unsure the changes I might have to make to ensure that `_fit` works, could I possibly try to extend `_HeterogenousMetaEstimator`? I have read `HierarchyEnsembleForecaster`'s internal functions to assist with `get_params` and I think it might be easier for me to implement them for this forecaster!
fkiraly: > could I possibly try to extend `_HeterogenousMetaEstimator`

That's what I meant! Sorry if my phrasing was confusing as to the precise architecture.

I meant, the `HierarchyEnsembleForecaster` is a good example on how `_HeterogenousMetaEstimator` is meant to be used as an intermediate base class.
shlok191: Ah, got it! I thought I’d confirm. As a short update, I’ve completed the implementation locally and just have some bugs to fix to pass all the test cases. This should be done soon!
fkiraly: ready for review?
shlok191: @fkiraly, yes! I have tried to follow the lead from `HierarchyEnsembleForecaster` to implement the necessary functionality to ensure get_params and other related functions work as expected.
shlok191: @fkiraly, I have added in the changes we discussed! I have added tests to ensure that the parameters return from `get_params()` are as we hope they should be and that the `name` in the `(name, object)` tuples is are always strings.

Additionally, I have renamed the forecaster to `TransformSelectForecaster` and I am dropping a question in dev-chat about the update!

Patch:
diff --git a/docs/source/api_reference/forecasting.rst b/docs/source/api_reference/forecasting.rst
index 1b03d86a6..4b6ca0254 100644
--- a/docs/source/api_reference/forecasting.rst
+++ b/docs/source/api_reference/forecasting.rst
@@ -50,6 +50,7 @@ Pipelines can also be constructed using ``*``, ``+``, and ``|`` dunders.
     ForecastByLevel
     Permute
     HierarchyEnsembleForecaster
+    TransformSelectForecaster
     FhPlexForecaster
     IgnoreX
     FallbackForecaster
diff --git a/extension_templates/alignment.py b/extension_templates/alignment.py
index c802067e6..2f33b27b1 100644
--- a/extension_templates/alignment.py
+++ b/extension_templates/alignment.py
@@ -101,6 +101,7 @@ class MyAligner(BaseAligner):
         "capability:multiple-alignment": False,  # can align more than two sequences?
         "capability:distance": False,  # does compute/return overall distance?
         "capability:distance-matrix": False,  # does compute/return distance matrix?
+        "capability:unequal_length": True,  # can align sequences of unequal length?
     }
 
     # todo: add any hyper-parameters and components to constructor
diff --git a/pyproject.toml b/pyproject.toml
index cbb0ffd6f..637463eed 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -80,7 +80,7 @@ all_extras = [
   "arch>=5.6,<7.1.0",
   "cloudpickle",
   "dash!=2.9.0",
-  "dask<2024.5.1; extra == 'dataframe'",
+  "dask<2024.5.2; extra == 'dataframe'",
   "dtw-python",
   'esig==0.9.7; python_version < "3.10"',
   'filterpy>=1.4.5; python_version < "3.11"',
@@ -121,7 +121,7 @@ all_extras_pandas2 = [
   "arch>=5.6,<7.1.0",
   "cloudpickle",
   "dash!=2.9.0",
-  "dask<2024.5.1; extra == 'dataframe'",
+  "dask<2024.5.2; extra == 'dataframe'",
   "dtw-python",
   'esig==0.9.7; python_version < "3.10"',
   'filterpy>=1.4.5; python_version < "3.11"',
diff --git a/sktime/alignment/base.py b/sktime/alignment/base.py
index 111866835..27873ed8d 100644
--- a/sktime/alignment/base.py
+++ b/sktime/alignment/base.py
@@ -39,6 +39,7 @@ class BaseAligner(BaseEstimator):
         "capability:multiple-alignment": False,  # can align more than two sequences?
         "capability:distance": False,  # does compute/return overall distance?
         "capability:distance-matrix": False,  # does compute/return distance matrix?
+        "capability:unequal_length": True,  # can align sequences of unequal length?
         "alignment_type": "full",  # does the aligner produce full or partial alignment
         "X_inner_mtype": "df-list",  # mtype of X expected by _fit
     }
@@ -71,13 +72,16 @@ class BaseAligner(BaseEstimator):
         # if fit is called, estimator is reset, including fitted state
         self.reset()
 
+        METADATA_TO_QUERY = ["is_equal_length", "n_instances"]
         valid, msg, X_metadata = check_is_scitype(
-            X, scitype="Panel", return_metadata=[], var_name="X"
+            X, scitype="Panel", return_metadata=METADATA_TO_QUERY, var_name="X"
         )
 
         if not valid:
             raise TypeError(msg)
 
+        self._check_capabilities(X_metadata)
+
         X_mtype = X_metadata["mtype"]
         X_inner_mtype = self.get_tag("X_inner_mtype")
 
@@ -351,3 +355,37 @@ class BaseAligner(BaseEstimator):
         distmat[1, 0] = dist
 
         return distmat
+
+    def _check_capabilities(self, X_metadata):
+        """Check if the aligner can align the input sequences.
+
+        Parameters
+        ----------
+        X_metadata : dict
+            metadata of the input sequences
+        """
+        # if aligner does not support unequal length sequences
+        # and X has unequal length, raise error
+        X_equal_length = X_metadata["is_equal_length"]
+        if not self.get_tag("capability:unequal_length", True) and not X_equal_length:
+            raise ValueError(
+                f"Aligner {self.__class__.__name__} instance does not support "
+                "alignment of unequal length sequences, but X passed "
+                "had unequal length. "
+                "Presence or lack of this capability may be depend on "
+                "hyper-parameters, especially for composites. Please consult the "
+                "documentation of the aligner for more information."
+            )
+
+        # if aligner does not support multiple alignment
+        # and X has more than two sequences, raise error
+        n_instances = X_metadata["n_instances"]
+        if not self.get_tag("capability:multiple-alignment", False) and n_instances > 2:
+            raise ValueError(
+                f"Aligner {self.__class__.__name__} instance does not support "
+                "alignment of multiple sequences, but X passed "
+                f"had {n_instances} sequences. "
+                "Presence or lack of this capability may be depend on "
+                "hyper-parameters, especially for composites. Please consult the "
+                "documentation of the aligner for more information."
+            )
diff --git a/sktime/alignment/dtw_numba.py b/sktime/alignment/dtw_numba.py
index d8ec2282b..d90148c8b 100644
--- a/sktime/alignment/dtw_numba.py
+++ b/sktime/alignment/dtw_numba.py
@@ -126,6 +126,7 @@ class AlignerDtwNumba(BaseAligner):
         "capability:multiple-alignment": False,  # can align more than two sequences?
         "capability:distance": True,  # does compute/return overall distance?
         "capability:distance-matrix": True,  # does compute/return distance matrix?
+        "capability:unequal_length": False,  # can align sequences of unequal length?
         "X_inner_mtype": "numpy3D",
     }
 
diff --git a/sktime/alignment/dtw_python.py b/sktime/alignment/dtw_python.py
index 45c8f9b9b..fd0c2a296 100644
--- a/sktime/alignment/dtw_python.py
+++ b/sktime/alignment/dtw_python.py
@@ -57,6 +57,7 @@ class AlignerDTW(BaseAligner):
         "capability:multiple-alignment": False,  # can align more than two sequences?
         "capability:distance": True,  # does compute/return overall distance?
         "capability:distance-matrix": True,  # does compute/return distance matrix?
+        "capability:unequal_length": True,  # can align sequences of unequal length?
         "alignment_type": "partial",
     }
 
diff --git a/sktime/alignment/edit_numba.py b/sktime/alignment/edit_numba.py
index dea505454..542cb2bf8 100644
--- a/sktime/alignment/edit_numba.py
+++ b/sktime/alignment/edit_numba.py
@@ -122,6 +122,7 @@ class AlignerEditNumba(BaseAligner):
         "capability:multiple-alignment": False,  # can align more than two sequences?
         "capability:distance": True,  # does compute/return overall distance?
         "capability:distance-matrix": True,  # does compute/return distance matrix?
+        "capability:unequal_length": False,  # can align sequences of unequal length?
         "alignment_type": "partial",
         "X_inner_mtype": "numpy3D",
     }
diff --git a/sktime/alignment/lucky.py b/sktime/alignment/lucky.py
index ea03ba79b..c159dc6c5 100644
--- a/sktime/alignment/lucky.py
+++ b/sktime/alignment/lucky.py
@@ -38,6 +38,7 @@ class AlignerLuckyDtw(BaseAligner):
         "capability:multiple-alignment": False,  # can align more than two sequences?
         "capability:distance": True,  # does compute/return overall distance?
         "capability:distance-matrix": True,  # does compute/return distance matrix?
+        "capability:unequal_length": True,  # can align sequences of unequal length?
         "alignment_type": "full",  # does the aligner produce full or partial alignment
     }
 
diff --git a/sktime/alignment/naive.py b/sktime/alignment/naive.py
index 7faf8fbad..f16c1d51b 100644
--- a/sktime/alignment/naive.py
+++ b/sktime/alignment/naive.py
@@ -31,6 +31,7 @@ class AlignerNaive(BaseAligner):
         # estimator type
         # --------------
         "capability:multiple-alignment": True,  # can align more than two sequences?
+        "capability:unequal_length": True,  # can align sequences of unequal length?
     }
 
     def __init__(self, strategy="start-end"):
diff --git a/sktime/dists_kernels/compose_from_align.py b/sktime/dists_kernels/compose_from_align.py
index 07790ad97..96d33f088 100644
--- a/sktime/dists_kernels/compose_from_align.py
+++ b/sktime/dists_kernels/compose_from_align.py
@@ -29,6 +29,9 @@ class DistFromAligner(BasePairwiseTransformerPanel):
 
         super().__init__()
 
+        if aligner is not None:
+            self.clone_tags(aligner, "capability:unequal_length")
+
     def _transform(self, X, X2=None):
         """Compute distance/kernel matrix.
 
diff --git a/sktime/forecasting/compose/__init__.py b/sktime/forecasting/compose/__init__.py
index 30b0c1971..f837af425 100644
--- a/sktime/forecasting/compose/__init__.py
+++ b/sktime/forecasting/compose/__init__.py
@@ -6,6 +6,7 @@ __all__ = [
     "HierarchyEnsembleForecaster",
     "ColumnEnsembleForecaster",
     "EnsembleForecaster",
+    "TransformSelectForecaster",
     "FallbackForecaster",
     "AutoEnsembleForecaster",
     "TransformedTargetForecaster",
@@ -63,3 +64,6 @@ from sktime.forecasting.compose._reduce import (
     make_reduction,
 )
 from sktime.forecasting.compose._stack import StackingForecaster
+from sktime.forecasting.compose._transform_select_forecaster import (
+    TransformSelectForecaster,
+)
diff --git a/sktime/forecasting/compose/_reduce.py b/sktime/forecasting/compose/_reduce.py
index 712bada3a..a9d750f66 100644
--- a/sktime/forecasting/compose/_reduce.py
+++ b/sktime/forecasting/compose/_reduce.py
@@ -1736,7 +1736,7 @@ def _create_fcst_df(target_date, origin_df, fill=None):
     else:
         values = fill
 
-    res = pd.DataFrame(values, index=index, columns=columns)
+    res = pd.DataFrame(values, index=index, columns=columns, dtype="float64")
 
     if isinstance(origin_df, pd.Series) and not isinstance(index, pd.MultiIndex):
         res = res.iloc[:, 0]
diff --git a/sktime/forecasting/compose/_transform_select_forecaster.py b/sktime/forecasting/compose/_transform_select_forecaster.py
new file mode 100644
index 000000000..bf7d2435f
--- /dev/null
+++ b/sktime/forecasting/compose/_transform_select_forecaster.py
@@ -0,0 +1,515 @@
+# copyright: sktime developers, BSD-3-Clause License (see LICENSE file)
+"""Implements a compositor that selects an ideal forecaster from a given transformer."""
+
+__author__ = ["shlok191"]
+
+from sktime.base._meta import _HeterogenousMetaEstimator
+from sktime.forecasting.base import BaseForecaster
+from sktime.forecasting.croston import Croston
+from sktime.forecasting.naive import NaiveForecaster
+from sktime.forecasting.trend import PolynomialTrendForecaster
+from sktime.transformations.base import BaseTransformer
+from sktime.transformations.series.adi_cv import ADICVTransformer
+
+
+class TransformSelectForecaster(BaseForecaster, _HeterogenousMetaEstimator):
+    """Compositor that utilizes varying forecasters by time series data's nature.
+
+    Applies a series-to-primitives transformer on a given time series. Based on the
+    generated value from the transformer, one of multiple forecasters provided by
+    the user in the form of a dictionary (key => category, value => forecaster) is
+    selected. Finally, the chosen forecaster is fit to the data for future predictions.
+
+    Parameters
+    ----------
+    forecasters : dict[sktime forecasters]
+        dict of forecasters with the key corresponding to categories generated
+        by the given transformer and the value corresponding to a sktime forecaster.
+
+    transformer : sktime transformer, default = ADICVTransformer()
+        A series-to-primitives sk-time transformer that generates a value
+        which can be used to quantify a choice of forecaster for the time series.
+
+        Note: To ensure correct functionality, the transformer must store the
+        generated category in the first column of the returned values when
+        the transform() or fit_transform() functions are called.
+
+    fallback_forecaster : sktime forecaster | None, Optional
+        A fallback forecaster that will be used if the category generated by
+        the transformer does not match any of the given forecasters.
+
+    Raises
+    ------
+    AssertionError: If a valid transformer (an instance of BaseTransformer)
+    is not passed or if valid forecasters (instances of BaseForecaster) are not given.
+
+    Examples
+    --------
+    This example showcases how the TransformSelectForecaster can be utilized to select
+    appropriate forecasters on the basis of the time series category determined by
+    the ADICVTransformer!
+
+    >>> from sktime.forecasting.compose._transform_select_forecaster import (
+    ...    TransformSelectForecaster)
+    >>> from sktime.forecasting.croston import Croston
+    >>> from sktime.forecasting.trend import PolynomialTrendForecaster
+    >>> from sktime.forecasting.naive import NaiveForecaster
+    >>> from sktime.transformations.series.adi_cv import ADICVTransformer
+
+    # Importing the methods which can generate data of specific categories
+    depending on their variance and average demand intervals.
+
+    >>> from sktime.transformations.series.tests.test_adi_cv import (
+    ...     _generate_erratic_series)
+
+    # The forecaster is defined which accepts a dictionary of forecasters,
+    a transformer and optionally a fallback_forecaster
+
+    >>> group_forecaster = TransformSelectForecaster(
+    ...     forecasters =
+    ...         {"smooth": NaiveForecaster(),
+    ...         "erratic": Croston(),
+    ...         "intermittent": PolynomialTrendForecaster()},
+    ...     transformer=ADICVTransformer(features=["class"]))
+
+    >>> generated_data = _generate_erratic_series()
+
+    # The fit function firstly passes the data through the given transformer
+    # to generate a given category. This category can be seen by the variable
+    # self.category_.
+
+    >>> group_forecaster = group_forecaster.fit(generated_data, fh=50)
+    >>> #print(f"The chosen category is: {group_forecaster.category}")
+
+    >>> # Print out the predicted value over the given forecasting horizon!
+    >>> # print(group_forecaster.predict(fh=50, X=None))
+    """
+
+    _tags = {
+        "y_inner_mtype": "pd.DataFrame",
+        "X_inner_mtype": "pd.DataFrame",
+        "scitype:y": "both",
+        "ignores-exogeneous-X": False,
+        "requires-fh-in-fit": False,
+        "enforce_index_type": None,
+        "authors": ["shlok191"],
+        "maintainers": ["shlok191"],
+        "python_version": None,
+    }
+
+    _steps_attr = "_forecasters"
+
+    def __init__(
+        self,
+        forecasters,
+        transformer=None,
+        fallback_forecaster=None,
+    ):
+        # saving arguments to object storage
+        if transformer is not None:
+            self.transformer = transformer
+
+        else:
+            self.transformer = ADICVTransformer(features=["class"])
+
+        self.forecasters = forecasters
+        self.fallback_forecaster = fallback_forecaster
+
+        self.transformer_ = self.transformer.clone()
+
+        super().__init__()
+
+        # validating passed arguments
+        assert isinstance(self.transformer_, BaseTransformer)
+
+        for forecaster in forecasters.values():
+            assert isinstance(forecaster, BaseForecaster)
+
+        # All checks OK!
+
+        # Assigning all capabilities on the basis of the capabilities
+        # of the passed forecasters
+        true_if_all_tags = {
+            "ignores-exogeneous-X": True,
+            "X-y-must-have-same-index": True,
+            "enforce_index_type": True,
+            "handles-missing-data": True,
+            "capability:insample": True,
+            "capability:pred_int": True,
+            "capability:pred_int:insample": True,
+        }
+
+        # Extrapolating values for flags that should be True if they are
+        # True for all of the given forecasters!
+        for tag in true_if_all_tags.keys():
+            # Checking the equivalent forecaster tags
+            true_for_all = True
+
+            for forecaster in self.forecasters.values():
+                # Fetching the forecaster tags
+                forecaster_tags = forecaster.get_tags()
+
+                if tag not in forecaster_tags or forecaster_tags[tag] is False:
+                    true_for_all = False
+                    break
+
+            # Perform this check for the fallback forecaster too
+            if fallback_forecaster is not None and (
+                tag not in fallback_forecaster.get_tags()
+                or fallback_forecaster.get_tags()[tag] is False
+            ):
+                true_for_all = False
+
+            true_if_all_tags[tag] = true_for_all
+
+        # Extrapolating values for flags that should be True if they are
+        # True for any of the given forecasters!
+        true_if_any_tags = {
+            "requires-fh-in-fit": True,
+            "X-y-must-have-same-index": True,
+        }
+
+        # Create a list of forecasters
+        forecasters = list(self.forecasters.items())
+
+        if self.fallback_forecaster is not None:
+            forecasters.append(("", self.fallback_forecaster))
+
+        # Update the tags by iterating through all tags
+        for tag in true_if_any_tags:
+            self._anytagis_then_set(tag, True, False, forecasters)
+
+        # Update the tags
+        self.set_tags(**true_if_all_tags)
+
+        # Finally, dynamically adding implementation of probabilistic
+        # functions depending on the tags set.
+        if self.get_tags()["capability:pred_int"]:
+            self._predict_interval = _predict_interval
+            self._predict_var = _predict_var
+            self._predict_proba = _predict_proba
+
+    def _fit(self, y, X=None, fh=None):
+        """Fit forecaster to training data.
+
+        private _fit containing the core logic, called from fit
+
+        For the _fit function to work as intended, the transformer
+        must generate and store the extrapolated category in the
+        first column.
+
+        Writes to self:
+            Sets fitted model attributes ending in "_".
+
+        Parameters
+        ----------
+        y : Pd.Series
+            The target time series to which we fit the data.
+
+        fh : ForecastingHorizon | None, optional (default=None)
+            The forecasting horizon with the steps ahead to predict.
+
+        X : Pd.Series | None, optional (default=None)
+            No exogenous variables are used for this.
+
+        Returns
+        -------
+        self : reference to self
+
+        Raises
+        ------
+        ValueError: If the extrapolated category has no provided forecaster
+        and if there is no fallback forecaster provided to the object!
+
+        Example:
+
+        If the passed transformer is an ADICVTransformer(), and the generated
+        series is a lumpy series; however, if there is no key matching "lumpy"
+        in the forecasters parameter, the fallback_forecaster will be used.
+        Additionally, if the fallback_forecaster is None, a ValueError will be thrown.
+        """
+        # passing time series through the provided transformer!
+
+        self.category_ = self.transformer_.fit_transform(X=y, y=X).iloc[0, 0]
+
+        # check if we have an available forecaster
+        if self.category_ not in self.forecasters:
+            if self.fallback_forecaster is None:
+                raise ValueError(
+                    "Forecaster not provided for given"
+                    + f"time series of type {self.category_}"
+                    + "and no fallback forecaster provided to use for this case."
+                )
+
+            # Adopt the fallback forecaster if possible
+            else:
+                self.chosen_forecaster_ = self.fallback_forecaster.clone()
+
+        else:
+            self.chosen_forecaster_ = self.forecasters[self.category_].clone()
+
+        # fitting the forecaster!
+        self.chosen_forecaster_.fit(y=y, X=X, fh=fh)
+
+        return self
+
+    def _predict(self, fh, X):
+        """Forecast time series at future horizon.
+
+        private _predict containing the core logic, called from predict
+
+        State required:
+            Requires state to be "fitted".
+
+        Accesses in self:
+            Fitted model attributes ending in "_"
+            self.cutoff
+
+        Parameters
+        ----------
+        fh : guaranteed to be ForecastingHorizon or None, optional (default=None)
+            The forecasting horizon with the steps ahead to to predict.
+            If not passed in _fit, guaranteed to be passed here
+
+        X : sktime time series object, optional (default=None)
+            guaranteed to be of an mtype in self.get_tag("X_inner_mtype")
+            Exogeneous time series for the forecast
+
+        Returns
+        -------
+        y_pred : sktime time series object
+            should be of the same type as seen in _fit, as in "y_inner_mtype" tag
+            Point predictions
+        """
+        # Obtain the prediction values for the given horizon.
+        y_pred = self.chosen_forecaster_.predict(fh=fh, X=X)
+
+        return y_pred
+
+    def _update(self, y, X=None, update_params=True):
+        """Update time series to incremental training data.
+
+        Does not update the extrapolated category from the given
+        transformer, and thus the chosen forecaster remains the same.
+
+        State required:
+            Requires state to be "fitted".
+
+        Accesses in self:
+            Fitted model attributes ending in "_"
+            self.cutoff
+
+        Writes to self:
+            Sets fitted model attributes ending in "_", if update_params=True.
+            Does not write to self if update_params=False.
+
+        Parameters
+        ----------
+        y : sktime time series object
+            guaranteed to be of an mtype in self.get_tag("y_inner_mtype")
+            Time series with which to update the forecaster.
+            if self.get_tag("scitype:y")=="univariate":
+                guaranteed to have a single column/variable
+            if self.get_tag("scitype:y")=="multivariate":
+                guaranteed to have 2 or more columns
+            if self.get_tag("scitype:y")=="both": no restrictions apply
+        X :  sktime time series object, optional (default=None)
+            guaranteed to be of an mtype in self.get_tag("X_inner_mtype")
+            Exogeneous time series for the forecast
+        update_params : bool, optional (default=True)
+            whether model parameters should be updated
+
+        Returns
+        -------
+        self : reference to self
+        """
+        self.chosen_forecaster_.update(y=y, X=X, update_params=update_params)
+
+    @classmethod
+    def get_test_params(cls, parameter_set="default"):
+        """Return testing parameter settings for the estimator.
+
+        Parameters
+        ----------
+        parameter_set : str, default="default"
+            Name of the set of test parameters to return, for use in tests. If no
+            special parameters are defined for a value, will return `"default"` set.
+            There are currently no reserved values for forecasters.
+
+        Returns
+        -------
+        params : dict or list of dict, default = {}
+            Parameters to create testing instances of the class
+            Each dict are parameters to construct an "interesting" test instance, i.e.,
+            `MyClass(**params)` or `MyClass(**params[i])` creates a valid test instance.
+            `create_test_instance` uses the first (or only) dictionary in `params`
+        """
+        param1 = {
+            "forecasters": {
+                "smooth": NaiveForecaster(),
+                "erratic": PolynomialTrendForecaster(),
+                "intermittent": Croston(),
+                "lumpy": NaiveForecaster(),
+            },
+            "transformer": ADICVTransformer(features=["class"]),
+            "fallback_forecaster": None,
+        }
+
+        # Attempting to utilize the fallback forecaster
+        param2 = {
+            "forecasters": {},
+            "transformer": ADICVTransformer(features=["class"]),
+            "fallback_forecaster": Croston(),
+        }
+
+        params = [param1, param2]
+        return params
+
+    @property
+    def _forecasters(self):
+        """Provides an internal list of the forecasters available.
+
+        Each list item is a tuple of the format (category, forecaster)
+        where the category for which the respective forecaster is chosen
+        and the forecaster itself as the values for each tuple.
+
+        Returns
+        -------
+        forecasters : list[tuple[str, strsktime forecasters]]
+            The list of forecasters which is returned. Also includes the
+            fallback forecaster with the category: "fallback_forecaster"
+        """
+        return list(self.forecasters.items()) + [
+            "fallback_forecaster",
+            self.fallback_forecaster,
+        ]
+
+    @_forecasters.setter
+    def _forecasters(self, new_forecasters):
+        """Provide new values for the forecasters.
+
+        Parameters
+        ----------
+        new_forecasters : list[tuple[str, strsktime forecasters]]
+            The list of new forecasters to update the object's forecasters with
+        """
+        # Accepting in possible new forecasters
+        for category, forecaster in new_forecasters:
+            # We assign this in a different way
+            if category != "fallback_forecaster":
+                self.forecasters[category] = forecaster
+
+            else:
+                self.fallback_forecaster = forecaster
+
+
+# Function implementations that will be added dynamically
+# if the conditions are met. explained further above!
+def _predict_interval(self, fh, X, coverage):
+    """Compute/return prediction quantiles for a forecast.
+
+    private _predict_interval containing the core logic,
+        called from predict_interval and possibly predict_quantiles
+
+    State required:
+        Requires state to be "fitted".
+
+    Accesses in self:
+        Fitted model attributes ending in "_"
+        self.cutoff
+
+    Parameters
+    ----------
+    fh : guaranteed to be ForecastingHorizon
+        The forecasting horizon with the steps ahead to to predict.
+    X :  sktime time series object, optional (default=None)
+        guaranteed to be of an mtype in self.get_tag("X_inner_mtype")
+        Exogeneous time series for the forecast
+    coverage : list of float (guaranteed not None and floats in [0,1] interval)
+        nominal coverage(s) of predictive interval(s)
+
+    Returns
+    -------
+    pred_int : pd.DataFrame
+        Column has multi-index: first level is variable name from y in fit,
+            second level coverage fractions for which intervals were computed.
+                in the same order as in input `coverage`.
+            Third level is string "lower" or "upper", for lower/upper interval end.
+        Row index is fh, with additional (upper) levels equal to instance levels,
+            from y seen in fit, if y_inner_mtype is Panel or Hierarchical.
+        Entries are forecasts of lower/upper interval end,
+            for var in col index, at nominal coverage in second col index,
+            lower/upper depending on third col index, for the row index.
+            Upper/lower interval end forecasts are equivalent to
+            quantile forecasts at alpha = 0.5 - c/2, 0.5 + c/2 for c in coverage.
+    """
+    # Call this function for the chosen forecaster
+    return self.chosen_forecaster_.predict_interval(fh=fh, X=X, coverage=coverage)
+
+
+def _predict_var(self, fh, X=None, cov=False):
+    """Forecast variance at future horizon.
+
+    private _predict_var containing the core logic, called from predict_var
+
+    Parameters
+    ----------
+    fh : guaranteed to be ForecastingHorizon or None, optional (default=None)
+        The forecasting horizon with the steps ahead to to predict.
+        If not passed in _fit, guaranteed to be passed here
+    X :  sktime time series object, optional (default=None)
+        guaranteed to be of an mtype in self.get_tag("X_inner_mtype")
+        Exogeneous time series for the forecast
+    cov : bool, optional (default=False)
+        if True, computes covariance matrix forecast.
+        if False, computes marginal variance forecasts.
+
+    Returns
+    -------
+    pred_var : pd.DataFrame, format dependent on `cov` variable
+        If cov=False:
+            Column names are exactly those of `y` passed in `fit`/`update`.
+                For nameless formats, column index will be a RangeIndex.
+            Row index is fh, with additional levels equal to instance levels,
+                from y seen in fit, if y_inner_mtype is Panel or Hierarchical.
+            Entries are variance forecasts, for var in col index.
+            A variance forecast for given variable and fh index is a predicted
+                variance for that variable and index, given observed data.
+        If cov=True:
+            Column index is a multiindex: 1st level is variable names (as above)
+                2nd level is fh.
+            Row index is fh, with additional levels equal to instance levels,
+                from y seen in fit, if y_inner_mtype is Panel or Hierarchical.
+            Entries are (co-)variance forecasts, for var in col index, and
+                covariance between time index in row and col.
+            Note: no covariance forecasts are returned between different variables.
+    """
+    return self.chosen_forecaster_.predict_var(fh=fh, X=X, cov=cov)
+
+
+def _predict_proba(self, fh, X, marginal=True):
+    """Compute/return fully probabilistic forecasts.
+
+    private _predict_proba containing the core logic, called from predict_proba
+
+    Parameters
+    ----------
+    fh : int, list, np.array or ForecastingHorizon (not optional)
+        The forecasting horizon encoding the time stamps to forecast at.
+        if has not been passed in fit, must be passed, not optional
+    X : sktime time series object, optional (default=None)
+            Exogeneous time series for the forecast
+        Should be of same scitype (Series, Panel, or Hierarchical) as y in fit
+        if self.get_tag("X-y-must-have-same-index"),
+            X.index must contain fh.index and y.index both
+    marginal : bool, optional (default=True)
+        whether returned distribution is marginal by time index
+
+    Returns
+    -------
+    pred_dist : sktime BaseDistribution
+        predictive distribution
+        if marginal=True, will be marginal distribution by time point
+        if marginal=False and implemented by method, will be joint
+    """
+    return self.chosen_forecaster_.predict_proba(fh=fh, X=X, marginal=marginal)
diff --git a/sktime/performance_metrics/forecasting/_classes.py b/sktime/performance_metrics/forecasting/_classes.py
index a8d358bb9..0914e3232 100644
--- a/sktime/performance_metrics/forecasting/_classes.py
+++ b/sktime/performance_metrics/forecasting/_classes.py
@@ -103,18 +103,23 @@ class BaseForecastingErrorMetric(BaseMetric):
 
     Parameters
     ----------
-    multioutput : {'raw_values', 'uniform_average'}  or array-like of shape \
+    multioutput : {'raw_values', 'uniform_average'} or array-like of shape \
             (n_outputs,), default='uniform_average'
         Defines whether and how to aggregate metric for across variables.
-        If 'uniform_average' (default), errors are mean-averaged across variables.
-        If array-like, errors are weighted averaged across variables, values as weights.
-        If 'raw_values', does not average errors across variables, columns are retained.
+
+        * If 'uniform_average' (default), errors are mean-averaged across variables.
+        * If array-like, errors are weighted averaged across variables,
+          values as weights.
+        * If 'raw_values', does not average errors across variables,
+          columns are retained.
 
     multilevel : {'raw_values', 'uniform_average', 'uniform_average_time'}
         Defines how to aggregate metric for hierarchical data (with levels).
-        If 'uniform_average' (default), errors are mean-averaged across levels.
-        If 'uniform_average_time', metric is applied to all data, ignoring level index.
-        If 'raw_values', does not average errors across levels, hierarchy is retained.
+
+        * If 'uniform_average' (default), errors are mean-averaged across levels.
+        * If 'uniform_average_time', metric is applied to all data,
+          ignoring level index.
+        * If 'raw_values', does not average errors across levels, hierarchy is retained.
     """
 
     _tags = {
@@ -141,31 +146,75 @@ class BaseForecastingErrorMetric(BaseMetric):
 
         Parameters
         ----------
-        y_true : time series in sktime compatible data container format
-            Ground truth (correct) target values
-            y can be in one of the following formats:
-            Series scitype: pd.Series, pd.DataFrame, or np.ndarray (1D or 2D)
-            Panel scitype: pd.DataFrame with 2-level row MultiIndex,
-                3D np.ndarray, list of Series pd.DataFrame, or nested pd.DataFrame
-            Hierarchical scitype: pd.DataFrame with 3 or more level row MultiIndex
-        y_pred :time series in sktime compatible data container format
-            Forecasted values to evaluate
-            must be of same format as y_true, same indices and columns if indexed
+        y_true : time series in ``sktime`` compatible data container format.
+            Ground truth (correct) target values.
+
+            Individual data formats in ``sktime`` are so-called :term:`mtype`
+            specifications, each mtype implements an abstract :term:`scitype`.
+
+            * ``Series`` scitype = individual time series, vanilla forecasting.
+              ``pd.DataFrame``, ``pd.Series``, or ``np.ndarray`` (1D or 2D)
+
+            * ``Panel`` scitype = collection of time series, global/panel forecasting.
+              ``pd.DataFrame`` with 2-level row ``MultiIndex`` ``(instance, time)``,
+              ``3D np.ndarray`` ``(instance, variable, time)``,
+              ``list`` of ``Series`` typed ``pd.DataFrame``
+
+            * ``Hierarchical`` scitype = hierarchical collection, for
+              hierarchical forecasting. ``pd.DataFrame`` with 3 or more level row
+              ``MultiIndex`` ``(hierarchy_1, ..., hierarchy_n, time)``
+
+            For further details on data format, see glossary on :term:`mtype`.
+            For usage, see forecasting tutorial ``examples/01_forecasting.ipynb``
+
+        y_pred : time series in ``sktime`` compatible data container format
+            Predicted values to evaluate against ground truth.
+            Must be of same format as ``y_true``, same indices and columns if indexed.
+
+        y_pred_benchmark : optional, time series in ``sktime`` compatible data container format  # noqa: E501
+            Benchmark predictions to compare ``y_pred`` to, used for relative metrics.
+            Required only if metric requires benchmark predictions,
+            as indicated by tag ``requires-y-pred-benchmark``.
+            Otherwise, can be passed to ensure interface consistency, but is ignored.
+            Must be of same format as ``y_true``, same indices and columns if indexed.
+
+        y_train : optional, time series in ``sktime`` compatible data container format
+            Training data used to normalize the error metric.
+            Required only if metric requires training data,
+            as indicated by tag ``requires-y-train``.
+            Otherwise, can be passed to ensure interface consistency, but is ignored.
+            Must be of same format as ``y_true``, same columns if indexed,
+            but not necessarily same indices.
+
+        sample_weight : optional, 1D array-like, default=None
+            Sample weights for each time point.
+
+            * If ``None``, the time indices are considered equally weighted.
+            * If an array, must be 1D.
+              If ``y_true`` and ``y_pred``are a single time series,
+              ``sample_weight`` must be of the same length as ``y_true``.
+              If the time series are panel or hierarchical, the length of all
+              individual time
+              series must be the same, and equal to the length of ``sample_weight``,
+              for all instances of time series passed.
 
         Returns
         -------
         loss : float, np.ndarray, or pd.DataFrame
             Calculated metric, averaged or by variable.
-            float if self.multioutput="uniform_average" or array-like
-                and self.multilevel="uniform_average" or "uniform_average_time"
-                value is metric averaged over variables and levels (see class docstring)
-            np.ndarray of shape (y_true.columns,) if self.multioutput="raw_values"
-                and self.multilevel="uniform_average" or "uniform_average_time"
-                i-th entry is metric calculated for i-th variable
-            pd.DataFrame if self.multilevel=raw.values
-                of shape (n_levels, ) if self.multioutput = "uniform_average" or array
-                of shape (n_levels, y_true.columns) if self.multioutput="raw_values"
-                metric is applied per level, row averaging (yes/no) as in multioutput
+            Weighted by ``sample_weight`` if provided.
+
+            * float if ``multioutput="uniform_average" or array-like,
+              and ``multilevel="uniform_average"`` or "uniform_average_time"``.
+              Value is metric averaged over variables and levels (see class docstring)
+            * ``np.ndarray`` of shape ``(y_true.columns,)``
+              if `multioutput="raw_values"``
+              and ``multilevel="uniform_average"`` or ``"uniform_average_time"``.
+              i-th entry is the, metric calculated for i-th variable
+            * ``pd.DataFrame`` if ``multilevel="raw_values"``.
+              of shape ``(n_levels, )``, if ``multioutput="uniform_average"``;
+              of shape ``(n_levels, y_true.columns)`` if ``multioutput="raw_values"``.
+              metric is applied per level, row averaging (yes/no) as in ``multioutput``.
         """
         return self.evaluate(y_true, y_pred, **kwargs)
 
@@ -174,31 +223,75 @@ class BaseForecastingErrorMetric(BaseMetric):
 
         Parameters
         ----------
-        y_true : time series in sktime compatible data container format
-            Ground truth (correct) target values
-            y can be in one of the following formats:
-            Series scitype: pd.Series, pd.DataFrame, or np.ndarray (1D or 2D)
-            Panel scitype: pd.DataFrame with 2-level row MultiIndex,
-                3D np.ndarray, list of Series pd.DataFrame, or nested pd.DataFrame
-            Hierarchical scitype: pd.DataFrame with 3 or more level row MultiIndex
-        y_pred :time series in sktime compatible data container format
-            Forecasted values to evaluate
-            must be of same format as y_true, same indices and columns if indexed
+        y_true : time series in ``sktime`` compatible data container format.
+            Ground truth (correct) target values.
+
+            Individual data formats in ``sktime`` are so-called :term:`mtype`
+            specifications, each mtype implements an abstract :term:`scitype`.
+
+            * ``Series`` scitype = individual time series, vanilla forecasting.
+              ``pd.DataFrame``, ``pd.Series``, or ``np.ndarray`` (1D or 2D)
+
+            * ``Panel`` scitype = collection of time series, global/panel forecasting.
+              ``pd.DataFrame`` with 2-level row ``MultiIndex`` ``(instance, time)``,
+              ``3D np.ndarray`` ``(instance, variable, time)``,
+              ``list`` of ``Series`` typed ``pd.DataFrame``
+
+            * ``Hierarchical`` scitype = hierarchical collection, for
+              hierarchical forecasting. ``pd.DataFrame`` with 3 or more level row
+              ``MultiIndex`` ``(hierarchy_1, ..., hierarchy_n, time)``
+
+            For further details on data format, see glossary on :term:`mtype`.
+            For usage, see forecasting tutorial ``examples/01_forecasting.ipynb``
+
+        y_pred : time series in ``sktime`` compatible data container format
+            Predicted values to evaluate against ground truth.
+            Must be of same format as ``y_true``, same indices and columns if indexed.
+
+        y_pred_benchmark : optional, time series in ``sktime`` compatible data container format  # noqa: E501
+            Benchmark predictions to compare ``y_pred`` to, used for relative metrics.
+            Required only if metric requires benchmark predictions,
+            as indicated by tag ``requires-y-pred-benchmark``.
+            Otherwise, can be passed to ensure interface consistency, but is ignored.
+            Must be of same format as ``y_true``, same indices and columns if indexed.
+
+        y_train : optional, time series in ``sktime`` compatible data container format
+            Training data used to normalize the error metric.
+            Required only if metric requires training data,
+            as indicated by tag ``requires-y-train``.
+            Otherwise, can be passed to ensure interface consistency, but is ignored.
+            Must be of same format as ``y_true``, same columns if indexed,
+            but not necessarily same indices.
+
+        sample_weight : optional, 1D array-like, default=None
+            Sample weights for each time point.
+
+            * If ``None``, the time indices are considered equally weighted.
+            * If an array, must be 1D.
+              If ``y_true`` and ``y_pred``are a single time series,
+              ``sample_weight`` must be of the same length as ``y_true``.
+              If the time series are panel or hierarchical, the length of all
+              individual time
+              series must be the same, and equal to the length of ``sample_weight``,
+              for all instances of time series passed.
 
         Returns
         -------
         loss : float, np.ndarray, or pd.DataFrame
             Calculated metric, averaged or by variable.
-            float if self.multioutput="uniform_average" or array-like
-                and self.multilevel="uniform_average" or "uniform_average_time"
-                value is metric averaged over variables and levels (see class docstring)
-            np.ndarray of shape (y_true.columns,) if self.multioutput="raw_values"
-                and self.multilevel="uniform_average" or "uniform_average_time"
-                i-th entry is metric calculated for i-th variable
-            pd.DataFrame if self.multilevel=raw.values
-                of shape (n_levels, ) if self.multioutput = "uniform_average" or array
-                of shape (n_levels, y_true.columns) if self.multioutput="raw_values"
-                metric is applied per level, row averaging (yes/no) as in multioutput
+            Weighted by ``sample_weight`` if provided.
+
+            * float if ``multioutput="uniform_average" or array-like,
+              and ``multilevel="uniform_average"`` or "uniform_average_time"``.
+              Value is metric averaged over variables and levels (see class docstring)
+            * ``np.ndarray`` of shape ``(y_true.columns,)``
+              if `multioutput="raw_values"``
+              and ``multilevel="uniform_average"`` or ``"uniform_average_time"``.
+              i-th entry is the, metric calculated for i-th variable
+            * ``pd.DataFrame`` if ``multilevel="raw_values"``.
+              of shape ``(n_levels, )``, if ``multioutput="uniform_average"``;
+              of shape ``(n_levels, y_true.columns)`` if ``multioutput="raw_values"``.
+              metric is applied per level, row averaging (yes/no) as in ``multioutput``.
         """
         multioutput = self.multioutput
         multilevel = self.multilevel
@@ -333,27 +426,70 @@ class BaseForecastingErrorMetric(BaseMetric):
 
         Parameters
         ----------
-        y_true : time series in sktime compatible data container format
-            Ground truth (correct) target values
-            y can be in one of the following formats:
-            Series scitype: pd.Series, pd.DataFrame, or np.ndarray (1D or 2D)
-            Panel scitype: pd.DataFrame with 2-level row MultiIndex,
-                3D np.ndarray, list of Series pd.DataFrame, or nested pd.DataFrame
-            Hierarchical scitype: pd.DataFrame with 3 or more level row MultiIndex
-        y_pred :time series in sktime compatible data container format
-            Forecasted values to evaluate
-            must be of same format as y_true, same indices and columns if indexed
+        y_true : time series in ``sktime`` compatible data container format.
+            Ground truth (correct) target values.
+
+            Individual data formats in ``sktime`` are so-called :term:`mtype`
+            specifications, each mtype implements an abstract :term:`scitype`.
+
+            * ``Series`` scitype = individual time series, vanilla forecasting.
+              ``pd.DataFrame``, ``pd.Series``, or ``np.ndarray`` (1D or 2D)
+
+            * ``Panel`` scitype = collection of time series, global/panel forecasting.
+              ``pd.DataFrame`` with 2-level row ``MultiIndex`` ``(instance, time)``,
+              ``3D np.ndarray`` ``(instance, variable, time)``,
+              ``list`` of ``Series`` typed ``pd.DataFrame``
+
+            * ``Hierarchical`` scitype = hierarchical collection, for
+              hierarchical forecasting. ``pd.DataFrame`` with 3 or more level row
+              ``MultiIndex`` ``(hierarchy_1, ..., hierarchy_n, time)``
+
+            For further details on data format, see glossary on :term:`mtype`.
+            For usage, see forecasting tutorial ``examples/01_forecasting.ipynb``
+
+        y_pred : time series in ``sktime`` compatible data container format
+            Predicted values to evaluate against ground truth.
+            Must be of same format as ``y_true``, same indices and columns if indexed.
+
+        y_pred_benchmark : optional, time series in ``sktime`` compatible data container format  # noqa: E501
+            Benchmark predictions to compare ``y_pred`` to, used for relative metrics.
+            Required only if metric requires benchmark predictions,
+            as indicated by tag ``requires-y-pred-benchmark``.
+            Otherwise, can be passed to ensure interface consistency, but is ignored.
+            Must be of same format as ``y_true``, same indices and columns if indexed.
+
+        y_train : optional, time series in ``sktime`` compatible data container format
+            Training data used to normalize the error metric.
+            Required only if metric requires training data,
+            as indicated by tag ``requires-y-train``.
+            Otherwise, can be passed to ensure interface consistency, but is ignored.
+            Must be of same format as ``y_true``, same columns if indexed,
+            but not necessarily same indices.
+
+        sample_weight : optional, 1D array-like, default=None
+            Sample weights for each time point.
+
+            * If ``None``, the time indices are considered equally weighted.
+            * If an array, must be 1D.
+              If ``y_true`` and ``y_pred``are a single time series,
+              ``sample_weight`` must be of the same length as ``y_true``.
+              If the time series are panel or hierarchical, the length of all
+              individual time
+              series must be the same, and equal to the length of ``sample_weight``,
+              for all instances of time series passed.
 
         Returns
         -------
         loss : pd.Series or pd.DataFrame
             Calculated metric, by time point (default=jackknife pseudo-values).
-            pd.Series if self.multioutput="uniform_average" or array-like
-                index is equal to index of y_true
-                entry at index i is metric at time i, averaged over variables
-            pd.DataFrame if self.multioutput="raw_values"
-                index and columns equal to those of y_true
-                i,j-th entry is metric at time i, at variable j
+            Weighted by ``sample_weight`` if provided.
+
+            * ``pd.Series`` if ``multioutput="uniform_average"`` or array-like.
+              index is equal to index of ``y_true``;
+              entry at index i is metric at time i, averaged over variables
+            * ``pd.DataFrame`` if ``multioutput="raw_values"``.
+              index and columns equal to those of ``y_true``;
+              i,j-th entry is metric at time i, at variable j
         """
         multioutput = self.multioutput
         multilevel = self.multilevel
@@ -558,6 +694,80 @@ class BaseForecastingErrorMetric(BaseMetric):
 
         return y_true, y_pred, multioutput, multilevel, kwargs
 
+    def _get_sample_weight(self, **kwargs):
+        """Get sample weights from kwargs.
+
+        Assumes that either ``sample_weight`` is passed, or not.
+        If ``sample_weight`` is passed, it is coerced to 1D numpy array and returned.
+        Otherwise, returns None.
+
+        Parameters
+        ----------
+        kwargs : dict
+            Dictionary of keyword arguments passed to the metric.
+
+        Returns
+        -------
+        sample_weight : 1D np.ndarray or None
+            1D numpy array of sample weights, or None if not passed.
+        """
+        sample_weight = kwargs.get("sample_weight", None)
+        if sample_weight is not None:
+            sample_weight = check_array(
+                sample_weight, ensure_2d=False, input_name="sample_weight"
+            )
+        return sample_weight
+
+    def _get_weighted_df(self, df, **kwargs):
+        """Get weighted DataFrame.
+
+        For n x m df, and kwargs containing sample_weight of length n,
+        returns df * sample_weight.reshape(-1, 1), i.e., weights
+        multiplied to each row of the DataFrame.
+
+        Parameters
+        ----------
+        df : pd.DataFrame
+            DataFrame to be weighted.
+
+        Returns
+        -------
+        df : pd.DataFrame
+            Weighted DataFrame.
+        """
+        sample_weight = self._get_sample_weight(**kwargs)
+        if sample_weight is not None:
+            df = df.mul(sample_weight, axis=0)
+        return df
+
+    def _handle_multioutput(self, df, multioutput):
+        """Handle multioutput parameter.
+
+        If multioutput is "raw_values", returns df unchanged.
+        If multioutput is "uniform_average", returns df.mean(axis=1) for pd.DataFrame,
+        or df.mean() for pd.Series.
+        If multioutput is array-like, returns df.dot(multioutput).
+
+        Parameters
+        ----------
+        df : pd.DataFrame or pd.Series
+            DataFrame to be handled, assumed result of metric calculation.
+        multioutput : str or array-like
+            Multioutput parameter.
+        """
+        if isinstance(multioutput, str):
+            if multioutput == "raw_values":
+                return df
+
+            if multioutput == "uniform_average":
+                if isinstance(df, pd.Series):
+                    return df.mean()
+                else:
+                    return df.mean(axis=1)
+
+        # else, we expect multioutput to be array-like
+        return df.dot(multioutput)
+
 
 class BaseForecastingErrorMetricFunc(BaseForecastingErrorMetric):
     """Adapter for numpy metrics."""
@@ -577,6 +787,13 @@ class BaseForecastingErrorMetricFunc(BaseForecastingErrorMetric):
         # adding kwargs to the metric, should not overwrite params (but does if clashes)
         params.update(kwargs)
 
+        # aliasing of sample_weight and horizon_weight keys
+        # this is for downwards compatibility with earlier sktime versions
+        if "sample_weight" in params.keys() and "horizon_weight" not in params.keys():
+            params["horizon_weight"] = params["sample_weight"]
+        if "horizon_weight" in params.keys() and "sample_weight" not in params.keys():
+            params["sample_weight"] = params["horizon_weight"]
+
         # calls class variable func, if available, or dynamic (object) variable
         # we need to call type since we store func as a class attribute
         if hasattr(type(self), "func") and isfunction(type(self).func):
@@ -1284,16 +1501,9 @@ class MeanAbsoluteError(BaseForecastingErrorMetric):
         multioutput = self.multioutput
 
         raw_values = (y_true - y_pred).abs()
+        raw_values = self._get_weighted_df(raw_values, **kwargs)
 
-        if isinstance(multioutput, str):
-            if multioutput == "raw_values":
-                return raw_values
-
-            if multioutput == "uniform_average":
-                return raw_values.mean(axis=1)
-
-        # else, we expect multioutput to be array-like
-        return raw_values.dot(multioutput)
+        return self._handle_multioutput(raw_values, multioutput)
 
 
 class MedianAbsoluteError(BaseForecastingErrorMetricFunc):
@@ -1556,20 +1766,13 @@ class MeanSquaredError(BaseForecastingErrorMetric):
         multioutput = self.multioutput
 
         raw_values = (y_true - y_pred) ** 2
+        raw_values = self._get_weighted_df(raw_values, **kwargs)
         msqe = raw_values.mean()
 
         if self.square_root:
             msqe = msqe.pow(0.5)
 
-        if isinstance(multioutput, str):
-            if multioutput == "raw_values":
-                return msqe
-
-            if multioutput == "uniform_average":
-                return msqe.mean()
-
-        # else, we expect multioutput to be array-like
-        return msqe.dot(multioutput)
+        return self._handle_multioutput(msqe, multioutput)
 
     def _evaluate_by_index(self, y_true, y_pred, **kwargs):
         """Return the metric evaluated at each time point.
@@ -1614,6 +1817,8 @@ class MeanSquaredError(BaseForecastingErrorMetric):
         else:
             pseudo_values = raw_values
 
+        pseudo_values = self._get_weighted_df(pseudo_values, **kwargs)
+
         if isinstance(multioutput, str):
             if multioutput == "raw_values":
                 return pseudo_values
diff --git a/sktime/registry/_tags.py b/sktime/registry/_tags.py
index 4964590b6..d5edd23be 100644
--- a/sktime/registry/_tags.py
+++ b/sktime/registry/_tags.py
@@ -865,6 +865,7 @@ class capability__unequal_length(_BaseTag):
     _tags = {
         "tag_name": "capability:unequal_length",
         "parent_type": [
+            "aligner",
             "classifier",
             "clusterer",
             "early_classifier",
diff --git a/sktime/transformations/series/impute.py b/sktime/transformations/series/impute.py
index 8089b8092..c924948c5 100644
--- a/sktime/transformations/series/impute.py
+++ b/sktime/transformations/series/impute.py
@@ -362,7 +362,7 @@ class Imputer(BaseTransformer):
                 )
 
                 # replace missing values with predicted values
-                X[col][na_index] = self._forecaster.predict(fh=fh, X=y)
+                X.loc[na_index, col] = self._forecaster.predict(fh=fh, X=y)
         return X
 
     @classmethod


Test patch:
diff --git a/sktime/forecasting/compose/tests/test_transformer_select_forecaster.py b/sktime/forecasting/compose/tests/test_transformer_select_forecaster.py
new file mode 100644
index 000000000..6887fd448
--- /dev/null
+++ b/sktime/forecasting/compose/tests/test_transformer_select_forecaster.py
@@ -0,0 +1,214 @@
+# copyright: sktime developers, BSD-3-Clause License (see LICENSE file).
+"""Tests for the TransformSelectForecaster"""
+
+import pytest
+
+from sktime.forecasting.compose import TransformSelectForecaster
+from sktime.forecasting.croston import Croston
+from sktime.forecasting.naive import NaiveForecaster
+from sktime.forecasting.trend import PolynomialTrendForecaster
+from sktime.transformations.series.adi_cv import ADICVTransformer
+from sktime.transformations.series.tests.test_adi_cv import (
+    _generate_erratic_series,
+    _generate_intermittent_series,
+    _generate_smooth_series,
+)
+
+
+@pytest.mark.parametrize(
+    "forecasters, series_generator, horizon, chosen_forecaster",
+    [
+        (
+            {
+                "smooth": NaiveForecaster(),
+                "intermittent": Croston(),
+            },
+            _generate_smooth_series,
+            50,
+            NaiveForecaster(),
+        ),
+        (
+            {
+                "erratic": PolynomialTrendForecaster(),
+                "lumpy": NaiveForecaster(),
+            },
+            _generate_erratic_series,
+            50,
+            PolynomialTrendForecaster(),
+        ),
+    ],
+)
+def test_forecaster_selection(
+    forecasters, series_generator, horizon, chosen_forecaster
+):
+    """Tests if the forecaster selects the correct forecasters on the basis
+    of the output from the transformer.
+
+    Parameters
+    ----------
+    forecasters : dict[sktime forecasters]
+        The forecasters provided by the user
+
+    series_generator : function
+        A function that generates test data
+
+    horizon : int
+        The forecasting horizon at which to predict values.
+
+    chosen_forecaster : sktime foreacsater
+       The forecaster that should be selected by the forecaster
+    """
+    # Defining the forecaster
+    forecaster = TransformSelectForecaster(forecasters=forecasters)
+    forecaster.fit(series_generator(), fh=horizon)
+
+    # Check if the type of the chosen forecaster matches the provided forecaster type
+    assert type(forecaster.chosen_forecaster_) is type(chosen_forecaster)
+
+
+@pytest.mark.parametrize(
+    "forecasters, series_generator, horizon, fallback_forecaster",
+    [
+        (
+            {
+                "smooth": NaiveForecaster(),
+                "intermittent": Croston(),
+            },
+            _generate_erratic_series,
+            50,
+            PolynomialTrendForecaster(),
+        ),
+        (
+            {
+                "erratic": PolynomialTrendForecaster(),
+                "lumpy": NaiveForecaster(),
+            },
+            _generate_smooth_series,
+            50,
+            None,
+        ),
+    ],
+)
+def test_fallback_forecaster(
+    forecasters, series_generator, horizon, fallback_forecaster
+):
+    """This function tests the validity of the fallback forecaster and what happens
+    if it is not present!
+
+    Parameters
+    ----------
+    forecasters : dict[sktime forecasters]
+        A dictionary consisting of sktime forecasters provided by the user
+
+    series_generator : function
+        A function that generates a time series of a particular type
+
+    horizon : int
+        The forecasting horizon at which to predict values.
+
+    fallback_forecaster : sktime foreacsater
+        A forecaster to choose if no options from `forecasters` is viable
+    """
+    # Creating our forecaster
+    forecaster = TransformSelectForecaster(
+        forecasters=forecasters, fallback_forecaster=fallback_forecaster
+    )
+
+    # Generating the required time series
+    y = series_generator()
+
+    if fallback_forecaster is None:
+        with pytest.raises(ValueError):
+            forecaster.fit(y, fh=horizon)
+
+    else:
+        forecaster.fit(y, fh=horizon)
+        assert type(forecaster.chosen_forecaster_) is type(fallback_forecaster)
+
+
+@pytest.mark.parametrize(
+    "forecasters, fallback_forecaster, transformer, horizon, series_generator, params",
+    [
+        (
+            {
+                "smooth": NaiveForecaster(),
+                "intermittent": Croston(),
+                "erratic": NaiveForecaster(),
+                "lumpy": PolynomialTrendForecaster(),
+            },
+            PolynomialTrendForecaster(),
+            ADICVTransformer(features=["class"]),
+            50,
+            _generate_intermittent_series,
+            {
+                "smooth": NaiveForecaster,
+                "intermittent": Croston,
+                "erratic": NaiveForecaster,
+                "lumpy": PolynomialTrendForecaster,
+            },
+        ),
+        (
+            {
+                "erratic": PolynomialTrendForecaster(),
+                "lumpy": NaiveForecaster(),
+            },
+            None,
+            ADICVTransformer(features=["class", "adi"]),
+            50,
+            _generate_erratic_series,
+            {
+                "erratic": PolynomialTrendForecaster,
+                "lumpy": NaiveForecaster,
+            },
+        ),
+    ],
+)
+def test_get_params(
+    forecasters,
+    fallback_forecaster,
+    transformer,
+    horizon,
+    series_generator,
+    params,
+):
+    """Tests the values generated from the get_params() function.
+
+    Parameters
+    ----------
+    forecasters : dict[sktime forecasters]
+        A dictionary consisting of sktime forecasters provided by the user
+
+    fallback_forecaster : sktime foreacsater
+        A forecaster to choose if no options from `forecasters` is viable
+
+    horizon : int
+        The forecasting horizon at which to predict values.
+
+    series_generator : function
+        A function that generates a time series of a particular type
+    """
+    # Creating our forecaster
+    forecaster = TransformSelectForecaster(
+        forecasters=forecasters,
+        transformer=transformer,
+        fallback_forecaster=fallback_forecaster,
+    )
+
+    # Fit the forecaster and then fetch the generated paramaters
+    forecaster.fit(y=series_generator(), fh=horizon)
+    forecaster_params = forecaster.get_params()
+
+    # Check that the forecasters map correctly in the params
+    for category, forecaster in forecaster_params["forecasters"].items():
+        assert type(forecaster) is params[category]
+
+    # Check the same for the fallback forecaster
+    if forecaster_params["fallback_forecaster"] is None:
+        assert fallback_forecaster is None
+
+    else:
+        assert type(forecaster_params["fallback_forecaster"] is fallback_forecaster)
+
+    # Check that the extrapolated values from the transformer are of type string
+    for category in forecaster_params.keys():
+        assert type(category) is str
diff --git a/sktime/performance_metrics/tests/test_metrics_classes.py b/sktime/performance_metrics/tests/test_metrics_classes.py
index 1eeda2719..09efc28cb 100644
--- a/sktime/performance_metrics/tests/test_metrics_classes.py
+++ b/sktime/performance_metrics/tests/test_metrics_classes.py
@@ -256,3 +256,26 @@ def test_uniform_average_time(metric):
     )
 
     assert np.allclose(res, res_noix)
+
+
+@pytest.mark.parametrize("metric", metrics, ids=names)
+def test_metric_weights(metric):
+    """Test that weights are correctly applied to the metric."""
+    y_true = np.array([3, -0.5, 2, 7, 2])
+    y_pred = np.array([2.5, 0.5, 2, 8, 2.25])
+    wts = np.array([0.1, 0.2, 0.1, 0.3, 2.4])
+
+    y_kwargs = {
+        "y_true": y_true,
+        "y_pred": y_pred,
+        "y_pred_benchmark": y_true,
+        "y_train": y_true,
+    }
+
+    metric_obj = metric()
+    if metric_obj(**y_kwargs) == metric_obj(sample_weight=wts, **y_kwargs):
+        raise ValueError(f"Metric {metric} does not handle sample_weight correctly")
+
+    # wt_metr = metric(sample_weight=wts)
+    # res_wt = wt_metr(y_true, y_pred)
+    # assert np.allclose(res_wt, metric_obj(y_true, y_pred, sample_weight=wts))