Repository: sktime/sktime
Pull request #7288 (2024-10-17 16:07:27)

[ENH] in forecasting metrics, allow a callable to be passed as `sample_weight` for dynamic weight generation
<!--
Welcome to sktime, and thanks for contributing!
Please have a look at our contribution guide:
https://www.sktime.net/en/latest/get_involved/contributing.html
-->

#### Reference Issues/PRs
<!--
Example: Fixes #1234. See also #3456.

Please use keywords (e.g., Fixes) to create link to the issues or pull requests
you resolved, so that they will automatically be closed when your pull request
is merged. See https://github.com/blog/1506-closing-issues-via-pull-requests.
If no issue exists, you can open one here: https://github.com/sktime/sktime/issues
-->
Addresses partially https://github.com/sktime/sktime/issues/7279 by allowing users to specify callables for `sample_weight`.

#### What does this implement/fix? Explain your changes.
<!--
A clear and concise description of what you have implemented.
-->

#### Does your contribution introduce a new dependency? If yes, which one?

<!--
Only relevant if you changed pyproject.toml.
We try to minimize dependencies in the core dependency set. There
are also further specific instructions to follow for soft dependencies.
See here for handling dependencies in sktime: https://www.sktime.net/en/latest/developer_guide/dependencies.html
-->

#### What should a reviewer concentrate their feedback on?

<!-- This section is particularly useful if you have a pull request that is still in development. You can guide the reviews to focus on the parts that are ready for their comments. We suggest using bullets (indicated by * or -) and filled checkboxes [x] here -->

#### Did you add any tests for the change?

<!-- This section is useful if you have added a test in addition to the existing ones. This will ensure that further changes to these files won't introduce the same kind of bug. It is considered good practice to add tests with newly added code to enforce the fact that the code actually works. This will reduce the chance of introducing logical bugs.
-->

#### Any other comments?
<!--
We value all user contributions, no matter how small or complex they are. If you have any questions, feel free to post
in the dev-chat channel on the sktime discord https://discord.com/invite/54ACzaFsn7. If we are slow to review (>3 working days), likewise feel free to ping us on discord. Thank you for your understanding during the review process.
-->

#### PR checklist
<!--
Please go through the checklist below. Please feel free to remove points if they are not applicable.
-->

##### 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)
- [ ] 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
- [ ] I've added the estimator to the API reference - in `docs/source/api_reference/taskname.rst`, follow the pattern.
- [ ] I've added one or more illustrative usage examples to the docstring, in a pydocstyle compliant `Examples` section.
- [ ] 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).

<!--
Thanks for contributing!
-->


MarkusSagen: @fkiraly Thank you for the feedback
I've started to adress some of the feedback now and included some tests

> the sample_weight_generator is still in the inits, I believe this change is not needed in the target design, so should be reverted.

Regarding this, I've now changed it a bit so metrics can still specify a sample weight generator and made one shared method for it in the base class, while still being scikit-learn compatible

As discussed in the issue thread, the reason for including it in the init method, is that usage in other closed methods (s.a. `sktime/forecasting/model_evaluation/_functions:evaluate`)
MarkusSagen: Great! I've reverted those changes and included it in another PR #7304 
MarkusSagen: Seems some of the tests failed because some of the CI/CD jobs failed to start
fkiraly: This is odd, it seems to have hit macos runners in the last days. Perhaps the queue generated by the python 3.13 release was too long and it got reset, as the macos runners tend to be slower than the rest. I restarted.
MarkusSagen: I've updated the docstrings now :) 

Patch:
diff --git a/sktime/performance_metrics/forecasting/_classes.py b/sktime/performance_metrics/forecasting/_classes.py
index d739d8057..5265e54bd 100644
--- a/sktime/performance_metrics/forecasting/_classes.py
+++ b/sktime/performance_metrics/forecasting/_classes.py
@@ -44,9 +44,12 @@ from sktime.performance_metrics.forecasting._functions import (
     median_squared_scaled_error,
     relative_loss,
 )
+from sktime.performance_metrics.forecasting.sample_weight._types import (
+    check_sample_weight_generator,
+)
 from sktime.utils.warnings import warn
 
-__author__ = ["mloning", "tch", "RNKuhns", "fkiraly"]
+__author__ = ["mloning", "tch", "RNKuhns", "fkiraly", "markussagen"]
 __all__ = [
     "make_forecasting_scorer",
     "MeanAbsoluteScaledError",
@@ -133,7 +136,11 @@ class BaseForecastingErrorMetric(BaseMetric):
         "reserved_params": ["multioutput", "multilevel"],
     }
 
-    def __init__(self, multioutput="uniform_average", multilevel="uniform_average"):
+    def __init__(
+        self,
+        multioutput="uniform_average",
+        multilevel="uniform_average",
+    ):
         self.multioutput = multioutput
         self.multilevel = multilevel
 
@@ -187,7 +194,7 @@ class BaseForecastingErrorMetric(BaseMetric):
             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_weight : optional, 1D array-like, or callable, default=None
             Sample weights for each time point.
 
             * If ``None``, the time indices are considered equally weighted.
@@ -198,6 +205,10 @@ class BaseForecastingErrorMetric(BaseMetric):
               individual time
               series must be the same, and equal to the length of ``sample_weight``,
               for all instances of time series passed.
+            * If a callable, it must follow ``SampleWeightGenerator`` interface,
+              or have one of the following signatures:
+              ``y_true: pd.DataFrame -> 1D array-like``,
+              or ``y_true: pd.DataFrame x y_pred: pd.DataFrame -> 1D array-like``.
 
         Returns
         -------
@@ -219,6 +230,18 @@ class BaseForecastingErrorMetric(BaseMetric):
         """  # noqa: E501
         return self.evaluate(y_true, y_pred, **kwargs)
 
+    def _apply_sample_weight_to_kwargs(self, y_true, y_pred, **kwargs):
+        """Apply sample weight to kwargs.
+
+        Sample weight is updated to kwargs if it is a callable and follows the
+        SampleWeightGenerator interface.
+        """
+        sample_weight = kwargs.get("sample_weight", None)
+        if callable(sample_weight) and check_sample_weight_generator(sample_weight):
+            kwargs["sample_weight"] = sample_weight(y_true, y_pred, **kwargs)
+
+        return kwargs
+
     def evaluate(self, y_true, y_pred, **kwargs):
         """Evaluate the desired metric on given inputs.
 
@@ -264,8 +287,8 @@ class BaseForecastingErrorMetric(BaseMetric):
             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.
+        sample_weight : optional, 1D array-like, or callable, default=None
+            Sample weights or callable for each time point.
 
             * If ``None``, the time indices are considered equally weighted.
             * If an array, must be 1D.
@@ -275,6 +298,10 @@ class BaseForecastingErrorMetric(BaseMetric):
               individual time
               series must be the same, and equal to the length of ``sample_weight``,
               for all instances of time series passed.
+            * If a callable, it must follow ``SampleWeightGenerator`` interface,
+              or have one of the following signatures:
+              ``y_true: pd.DataFrame -> 1D array-like``,
+              or ``y_true: pd.DataFrame x y_pred: pd.DataFrame -> 1D array-like``.
 
         Returns
         -------
@@ -296,11 +323,16 @@ class BaseForecastingErrorMetric(BaseMetric):
         """  # noqa: E501
         multioutput = self.multioutput
         multilevel = self.multilevel
+
         # Input checks and conversions
         y_true_inner, y_pred_inner, multioutput, multilevel, kwargs = self._check_ys(
             y_true, y_pred, multioutput, multilevel, **kwargs
         )
 
+        kwargs = self._apply_sample_weight_to_kwargs(
+            y_true=y_true_inner, y_pred=y_pred_inner, **kwargs
+        )
+
         requires_vectorization = isinstance(y_true_inner, VectorizedDF)
         if not requires_vectorization:
             # pass to inner function
@@ -467,8 +499,8 @@ class BaseForecastingErrorMetric(BaseMetric):
             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.
+        sample_weight : optional, 1D array-like, or callable, default=None
+            Sample weights or callable for each time point.
 
             * If ``None``, the time indices are considered equally weighted.
             * If an array, must be 1D.
@@ -478,6 +510,10 @@ class BaseForecastingErrorMetric(BaseMetric):
               individual time
               series must be the same, and equal to the length of ``sample_weight``,
               for all instances of time series passed.
+            * If a callable, it must follow ``SampleWeightGenerator`` interface,
+              or have one of the following signatures:
+              ``y_true: pd.DataFrame -> 1D array-like``,
+              or ``y_true: pd.DataFrame x y_pred: pd.DataFrame -> 1D array-like``.
 
         Returns
         -------
@@ -494,10 +530,16 @@ class BaseForecastingErrorMetric(BaseMetric):
         """  # noqa: E501
         multioutput = self.multioutput
         multilevel = self.multilevel
+
         # Input checks and conversions
         y_true_inner, y_pred_inner, multioutput, multilevel, kwargs = self._check_ys(
             y_true, y_pred, multioutput, multilevel, **kwargs
         )
+
+        kwargs = self._apply_sample_weight_to_kwargs(
+            y_true=y_true_inner, y_pred=y_pred_inner, **kwargs
+        )
+
         requires_vectorization = isinstance(y_true_inner, VectorizedDF)
         if not requires_vectorization:
             # pass to inner function
@@ -695,7 +737,7 @@ class BaseForecastingErrorMetric(BaseMetric):
 
         return y_true, y_pred, multioutput, multilevel, kwargs
 
-    def _get_sample_weight(self, **kwargs):
+    def _set_sample_weight_on_kwargs(self, **kwargs):
         """Get sample weights from kwargs.
 
         Assumes that either ``sample_weight`` is passed, or not.
@@ -736,7 +778,7 @@ class BaseForecastingErrorMetric(BaseMetric):
         df : pd.DataFrame
             Weighted DataFrame.
         """
-        sample_weight = self._get_sample_weight(**kwargs)
+        sample_weight = self._set_sample_weight_on_kwargs(**kwargs)
         if sample_weight is not None:
             df = df.mul(sample_weight, axis=0)
         return df
@@ -846,7 +888,6 @@ class _DynamicForecastingErrorMetric(BaseForecastingErrorMetricFunc):
         self.func = func
         self.name = name
         self.lower_is_better = lower_is_better
-
         super().__init__(multioutput=multioutput, multilevel=multilevel)
 
         self.set_tags(**{"lower_is_better": lower_is_better})
@@ -1581,13 +1622,6 @@ class MedianAbsoluteError(BaseForecastingErrorMetricFunc):
 
     func = median_absolute_error
 
-    def __init__(
-        self,
-        multioutput="uniform_average",
-        multilevel="uniform_average",
-    ):
-        super().__init__(multioutput=multioutput, multilevel=multilevel)
-
     def _evaluate_by_index(self, y_true, y_pred, **kwargs):
         """Return the metric evaluated at each time point.
 
diff --git a/sktime/performance_metrics/forecasting/probabilistic/_classes.py b/sktime/performance_metrics/forecasting/probabilistic/_classes.py
index 961a3b274..7aff5d314 100644
--- a/sktime/performance_metrics/forecasting/probabilistic/_classes.py
+++ b/sktime/performance_metrics/forecasting/probabilistic/_classes.py
@@ -55,7 +55,7 @@ class _BaseProbaForecastingErrorMetric(BaseForecastingErrorMetric):
                 (fh, n_outputs) where fh is the forecasting horizon
             Ground truth (correct) target values.
 
-        y_pred : return object of probabilistic predictition method scitype:y_pred
+        y_pred : return object of probabilistic prediction method scitype:y_pred
             must be at fh and for variables equal to those in y_true.
 
         Returns
@@ -82,7 +82,7 @@ class _BaseProbaForecastingErrorMetric(BaseForecastingErrorMetric):
                 (fh, n_outputs) where fh is the forecasting horizon
             Ground truth (correct) target values.
 
-        y_pred : return object of probabilistic predictition method scitype:y_pred
+        y_pred : return object of probabilistic prediction method scitype:y_pred
             must be at fh and for variables equal to those in y_true
 
         multioutput : string "uniform_average" or "raw_values" determines how\
@@ -176,7 +176,7 @@ class _BaseProbaForecastingErrorMetric(BaseForecastingErrorMetric):
                 (fh, n_outputs) where fh is the forecasting horizon
             Ground truth (correct) target values.
 
-        y_pred : return object of probabilistic predictition method scitype:y_pred
+        y_pred : return object of probabilistic prediction method scitype:y_pred
             must be at fh and for variables equal to those in y_true
 
         multioutput : string "uniform_average" or "raw_values" determines how\
@@ -855,7 +855,7 @@ class _BaseDistrForecastingMetric(_BaseProbaForecastingErrorMetric):
                 (fh, n_outputs) where fh is the forecasting horizon
             Ground truth (correct) target values.
 
-        y_pred : return object of probabilistic predictition method scitype:y_pred
+        y_pred : return object of probabilistic prediction method scitype:y_pred
             must be at fh and for variables equal to those in y_true
 
         multioutput : {'raw_values', 'uniform_average'} or array-like of shape \
diff --git a/sktime/performance_metrics/forecasting/sample_weight/__init__.py b/sktime/performance_metrics/forecasting/sample_weight/__init__.py
new file mode 100644
index 000000000..e7f69d293
--- /dev/null
+++ b/sktime/performance_metrics/forecasting/sample_weight/__init__.py
@@ -0,0 +1,16 @@
+"""Sample weight generators for sktime performance metrics."""
+
+__author__ = ["markussagen"]
+__all__ = [
+    "BaseSampleWeightGenerator",
+    "check_sample_weight_generator",
+    "SampleWeightGenerator",
+]
+
+from sktime.performance_metrics.forecasting.sample_weight._base import (
+    BaseSampleWeightGenerator,
+)
+from sktime.performance_metrics.forecasting.sample_weight._types import (
+    SampleWeightGenerator,
+    check_sample_weight_generator,
+)
diff --git a/sktime/performance_metrics/forecasting/sample_weight/_base.py b/sktime/performance_metrics/forecasting/sample_weight/_base.py
new file mode 100644
index 000000000..a2530baf0
--- /dev/null
+++ b/sktime/performance_metrics/forecasting/sample_weight/_base.py
@@ -0,0 +1,77 @@
+"""Implements base class for customer sample weights of performance metric in sktime."""
+
+__author__ = ["markussagen"]
+__all__ = ["BaseSampleWeightGenerator", "check_sample_weight_generator"]
+
+from sktime.base import BaseObject
+from sktime.performance_metrics.forecasting.sample_weight._types import (
+    check_sample_weight_generator,
+)
+
+
+class BaseSampleWeightGenerator(BaseObject):
+    """Base class for defining sample weight generators in sktime.
+
+    Extends sktime BaseObject.
+    """
+
+    _tags = {
+        "object_type": "sample_weight",
+        "authors": "markussagen",
+        "maintainers": "markussagen",
+    }
+
+    def __init__(self):
+        super().__init__()
+
+    def __call__(self, y_true, y_pred=None, **kwargs):
+        """Generate sample weights for a given set of true and predicted values.
+
+        Parameters
+        ----------
+        y_true : time series in sktime compatible data container format
+            Ground truth (correct) target values.
+
+        y_pred : time series in sktime compatible data container format, optional
+            Predicted values to evaluate against ground truth.
+            Must be of same format as y_true, same indices and columns if indexed.
+            If None, only y_true is used for weight generation.
+
+        **kwargs : dict
+            Additional keyword arguments specific to the weight generation method.
+
+        Returns
+        -------
+        sample_weight : 1D np.ndarray or None
+            Sample weights for each time point.
+
+            * 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.
+            * If None, the time indices are considered equally weighted.
+        """
+        raise NotImplementedError("abstract method")
+
+    @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.
+
+        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``
+        """
+        return [{}]
diff --git a/sktime/performance_metrics/forecasting/sample_weight/_types.py b/sktime/performance_metrics/forecasting/sample_weight/_types.py
new file mode 100644
index 000000000..9e5aeb93a
--- /dev/null
+++ b/sktime/performance_metrics/forecasting/sample_weight/_types.py
@@ -0,0 +1,113 @@
+"""Implements base class for customer sample weights of performance metric in sktime."""
+
+__author__ = ["markussagen"]
+__all__ = ["SampleWeightGenerator", "check_sample_weight_generator"]
+
+from inspect import signature
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class SampleWeightGenerator(Protocol):
+    """Protocol for sample weight generators.
+
+    This protocol defines the interface for sample weight generator functions used in
+    performance metrics calculations. Sample weight generators are used to assign
+    weights to individual time points in time series data.
+
+    The generator must have at least one parameter (y_true) and accept **kwargs.
+    It must return a 1D numpy array of sample weights.
+
+    Parameters
+    ----------
+    y_true : array-like of shape (n_samples,) or (n_samples, n_outputs)
+        Ground truth (correct) target values.
+    y_pred : array-like of shape (n_samples,) or (n_samples, n_outputs), optional
+        Estimated target values.
+    **kwargs : dict
+        Additional keyword arguments that may be used in weight calculation.
+
+    Returns
+    -------
+    sample_weights : np.ndarray of shape (n_samples,)
+        1D array of sample weights.
+
+    Raises
+    ------
+    ValueError
+        If the sample weight generator does not have at least one parameter (y_true).
+        If the first parameter is not y_true.
+        If the sample weight generator does not accept **kwargs.
+
+    Notes
+    -----
+    Implementations of this protocol should ensure that the returned weights are
+    non-negative and sum to a value greater than zero.
+
+    Examples
+    --------
+    >>> import numpy as np
+    >>> class ExampleWeightGenerator:
+    ...     def __call__(self, y_true, y_pred=None, **kwargs):
+    ...         return np.ones_like(y_true)
+    >>> weight_gen = ExampleWeightGenerator()
+    >>> y_true = np.array([1, 2, 3, 4, 5])
+    >>> weights = weight_gen(y_true)
+    >>> print(weights)
+    [1. 1. 1. 1. 1.]
+    """
+
+    def __call__(self, y_true, y_pred=None, **kwargs):
+        """Generate sample weights for time series data.
+
+        This method should be implemented by concrete sample weight generators.
+
+        Parameters
+        ----------
+        y_true : array-like of shape (n_samples,) or (n_samples, n_outputs)
+            Ground truth (correct) target values.
+        y_pred : array-like of shape (n_samples,) or (n_samples, n_outputs), optional
+            Estimated target values. Default is None.
+        **kwargs : dict
+            Additional keyword arguments that may be used in weight calculation.
+
+        Returns
+        -------
+        sample_weights : np.ndarray of shape (n_samples,)
+            1D array of sample weights.
+
+        Notes
+        -----
+        - The method should always return a 1D numpy array of weights.
+        - The length of the returned array should match the number of samples in y_true.
+        - If y_pred is not used in the weight calculation, it should be ignored.
+        """
+        ...
+
+
+def check_sample_weight_generator(obj):
+    """Check if obj is a valid SampleWeightGenerator."""
+    if obj is None or not callable(obj):
+        return False
+
+    sig = signature(obj)
+    params = sig.parameters
+
+    # Check if the function has at least one parameter (y_true)
+    if len(params) < 1:
+        msg = "Sample weight generator must have at least one parameter (y_true)"
+        raise ValueError(msg)
+
+    param_names = list(params.keys())
+
+    # Check if the first parameter is y_true
+    if param_names[0] != "y_true":
+        raise ValueError("First parameter of sample weight generator must be 'y_true'")
+
+    # Check if the function accepts **kwargs
+    if not any(param.kind == param.VAR_KEYWORD for param in params.values()):
+        raise ValueError("Sample weight generator must accept **kwargs")
+
+    # NOTE: this does not check the return type!
+
+    return True


Test patch:
diff --git a/sktime/performance_metrics/forecasting/sample_weight/tests/__init__.py b/sktime/performance_metrics/forecasting/sample_weight/tests/__init__.py
new file mode 100644
index 000000000..1af936133
--- /dev/null
+++ b/sktime/performance_metrics/forecasting/sample_weight/tests/__init__.py
@@ -0,0 +1,1 @@
+"""Tests for sktime sample weight generators."""
diff --git a/sktime/performance_metrics/forecasting/sample_weight/tests/test_sample_weight_generator.py b/sktime/performance_metrics/forecasting/sample_weight/tests/test_sample_weight_generator.py
new file mode 100644
index 000000000..4cd7e1ca0
--- /dev/null
+++ b/sktime/performance_metrics/forecasting/sample_weight/tests/test_sample_weight_generator.py
@@ -0,0 +1,260 @@
+"""Tests for sample weight generators."""
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from sktime.performance_metrics.forecasting.sample_weight._base import (
+    BaseSampleWeightGenerator,
+)
+from sktime.performance_metrics.forecasting.sample_weight._types import (
+    check_sample_weight_generator,
+)
+from sktime.tests.test_switch import run_test_module_changed
+
+
+class DummyWeightGenerator(BaseSampleWeightGenerator):
+    def __call__(self, y_true, y_pred=None, **kwargs) -> None:
+        return None
+
+
+class UniformWeightGenerator(BaseSampleWeightGenerator):
+    def __call__(self, y_true, y_pred=None, **kwargs) -> np.ndarray:
+        return np.ones_like(y_true)
+
+
+class RecencyWeightGenerator(BaseSampleWeightGenerator):
+    def __init__(self, decay_rate=0.1):
+        self.decay_rate = decay_rate
+
+    def __call__(self, y_true, y_pred=None, **kwargs):
+        weights = np.exp(-self.decay_rate * np.arange(len(y_true))[::-1])
+        return weights / np.sum(weights)
+
+
+class DateRangeWeightGenerator(BaseSampleWeightGenerator):
+    def __init__(self, date_ranges, default_weight=1.0):
+        self.date_ranges = date_ranges
+        self.default_weight = default_weight
+
+    def _get_dt(self, date_series):
+        """Get the datetime object from the series."""
+        dt = date_series
+        if isinstance(date_series, pd.Timestamp):
+            dt = date_series
+        elif isinstance(date_series, pd.DatetimeIndex):
+            dt = date_series
+        elif isinstance(date_series, pd.Series):
+            dt = date_series.dt
+        elif isinstance(date_series, pd.DataFrame):
+            dt = date_series.dt
+        elif isinstance(date_series, np.datetime64):
+            dt = pd.to_datetime(date_series)
+        else:
+            msg = f"Invalid type for conversion to datetime: {type(date_series)}"
+            raise ValueError(msg)
+        return dt
+
+    def __call__(self, y_true, y_pred=None, **kwargs):
+        if "dates" not in kwargs:
+            msg = "'dates' must be provided in kwargs for DateRangeWeightGenerator"
+            raise ValueError(msg)
+
+        if not (
+            isinstance(self.date_ranges, list)
+            and (
+                all(isinstance(range_config, dict) for range_config in self.date_ranges)
+            )
+        ):
+            raise ValueError("'date_ranges' must be a list of dictionaries")
+
+        if not all(
+            "start" in range_config
+            and "end" in range_config
+            and "weight" in range_config
+            for range_config in self.date_ranges
+        ):
+            msg = "'date_ranges' must be a list of dicts with the keys: "
+            msg += "'start', 'end', and 'weight'"
+            raise ValueError(msg)
+
+        dates = pd.to_datetime(kwargs["dates"])
+        weights = np.full(len(y_true), self.default_weight)
+
+        for range_config in self.date_ranges:
+            start_date = pd.to_datetime(range_config["start"], format="%m-%d")
+            end_date = pd.to_datetime(range_config["end"], format="%m-%d")
+
+            dt = self._get_dt(dates)
+            mask = (
+                (dt.month >= start_date.month)
+                & (dt.month <= end_date.month)
+                & (dt.day >= start_date.day)
+                & (dt.day <= end_date.day)
+            )
+            weights[mask] = range_config["weight"]
+
+        return weights / np.sum(weights)
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_base_sample_weight_generator():
+    """Test BaseSampleWeightGenerator."""
+
+    class TestGenerator(BaseSampleWeightGenerator):
+        def __call__(self, y_true, y_pred=None, **kwargs):
+            return np.ones_like(y_true)
+
+    generator = TestGenerator()
+    y_true = np.array([1, 2, 3])
+    weights = generator(y_true)
+
+    assert isinstance(weights, np.ndarray)
+    assert np.array_equal(weights, np.ones_like(y_true))
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_check_sample_weight_generator():
+    """Test check_sample_weight_generator function."""
+
+    def valid_generator(y_true, y_pred=None, **kwargs):
+        return np.ones_like(y_true)
+
+    def valid_generator_no_params(y_true, y_pred=None, **kwargs):
+        return None
+
+    def valid_generator_no_y_pred(y_true, **kwargs):
+        return np.ones_like(y_true)
+
+    assert check_sample_weight_generator(valid_generator)
+    assert check_sample_weight_generator(valid_generator_no_params)
+    assert check_sample_weight_generator(valid_generator_no_y_pred)
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_check_sample_weight_generator_are_invalid():
+    """Test check_sample_weight_generator function."""
+
+    def valid_generator_no_kwargs(y_true, y_pred=None):
+        return np.ones_like(y_true)
+
+    def invalid_generator():
+        return "invalid"
+
+    def invalid_generator_no_param_y_true(invalid):
+        return np.ones((1, 1))
+
+    # Validate
+    with pytest.raises(ValueError):
+        check_sample_weight_generator(valid_generator_no_kwargs)
+
+    with pytest.raises(ValueError):
+        check_sample_weight_generator(invalid_generator)
+
+    with pytest.raises(ValueError):
+        check_sample_weight_generator(invalid_generator_no_param_y_true)
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_dummy_weight_generator():
+    generator = DummyWeightGenerator()
+    y_true = np.array([1, 2, 3])
+    weights = generator(y_true)
+
+    assert weights is None
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_uniform_weight_generator():
+    generator = UniformWeightGenerator()
+    y_true = np.array([1, 2, 3])
+    weights = generator(y_true)
+
+    assert isinstance(weights, np.ndarray)
+    assert np.array_equal(weights, np.ones_like(y_true))
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_recency_weight_generator():
+    generator = RecencyWeightGenerator(decay_rate=0.1)
+    y_true = np.array([1, 2, 3])
+    weights = generator(y_true)
+
+    assert isinstance(weights, np.ndarray)
+    assert len(weights) == len(y_true)
+    assert np.all(weights[:-1] < weights[1:])  # Weights should be increasing
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_date_range_weight_generator():
+    date_ranges = [
+        {"start": "01-01", "end": "03-31", "weight": 2},
+        {"start": "06-01", "end": "08-31", "weight": 3},
+    ]
+    generator = DateRangeWeightGenerator(date_ranges=date_ranges)
+    y_true = np.array([1, 2, 3, 4, 5])
+    dates = pd.date_range(start="2023-01-01", periods=5, freq="MS")
+    weights = generator(y_true, dates=dates)
+
+    assert isinstance(weights, np.ndarray)
+    assert len(weights) == len(y_true)
+    assert weights[0] > weights[3]  # January should have higher weight than April
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_sample_weight_generator_with_y_pred():
+    class TestGenerator(BaseSampleWeightGenerator):
+        def __call__(self, y_true, y_pred=None, **kwargs):
+            if y_pred is None:
+                return np.ones_like(y_true)
+            return np.abs(y_true - y_pred)
+
+    generator = TestGenerator()
+    y_true = np.array([1, 2, 3])
+    y_pred = np.array([1.1, 2.2, 2.8])
+    weights = generator(y_true, y_pred)
+
+    assert isinstance(weights, np.ndarray)
+    assert np.allclose(weights, np.abs(y_true - y_pred))
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+def test_sample_weight_generator_with_kwargs():
+    class TestGenerator(BaseSampleWeightGenerator):
+        def __call__(self, y_true, y_pred=None, **kwargs):
+            factor = kwargs.get("factor", 1)
+            return np.ones_like(y_true) * factor
+
+    generator = TestGenerator()
+    y_true = np.array([1, 2, 3])
+    weights = generator(y_true, factor=2)
+
+    assert isinstance(weights, np.ndarray)
+    assert np.array_equal(weights, np.ones_like(y_true) * 2)
diff --git a/sktime/performance_metrics/tests/test_performance_metrics_forecasting.py b/sktime/performance_metrics/tests/test_performance_metrics_forecasting.py
index 5789c8121..63ff1d140 100644
--- a/sktime/performance_metrics/tests/test_performance_metrics_forecasting.py
+++ b/sktime/performance_metrics/tests/test_performance_metrics_forecasting.py
@@ -47,6 +47,9 @@ from sktime.performance_metrics.forecasting import (
     median_squared_scaled_error,
     relative_loss,
 )
+from sktime.performance_metrics.forecasting.sample_weight._base import (
+    BaseSampleWeightGenerator,
+)
 from sktime.performance_metrics.tests._config import RANDOM_SEED
 from sktime.tests.test_switch import run_test_module_changed
 from sktime.utils._testing.series import _make_series
@@ -588,3 +591,99 @@ def test_y_true_y_pred_inconsistent_n_variables_raises_error(metric_func_name):
         ValueError, match="y_true and y_pred have different number of output"
     ):
         metric_func(y_true, y_pred, y_train=y_train, y_pred_benchmark=y_pred_benchmark)
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+@pytest.mark.parametrize("random_state", RANDOM_STATES)
+@pytest.mark.parametrize("metric_func_name", LOSS_RESULTS.keys())
+def test_sample_weight_generator_has_no_effect_on_metric_function(
+    metric_func_name, random_state
+):
+    """Tests that loss function with univariate input should return scalar number."""
+    y = _make_series(n_timepoints=75, random_state=random_state)
+    y_train, y_true = y.iloc[:50], y.iloc[50:]
+    y_pred = y.shift(1).iloc[50:]
+    y_pred_benchmark = y.rolling(2).mean().iloc[50:]
+
+    class TestWeightGenerator(BaseSampleWeightGenerator):
+        def __call__(self, y_true, y_pred=None, **kwargs):
+            return np.arange(1, len(y_true) + 1)
+
+    weight_generator = TestWeightGenerator()
+
+    metric_func = LOSS_RESULTS[metric_func_name]["func"]
+
+    function_loss = metric_func(
+        y_true, y_pred, y_train=y_train, y_pred_benchmark=y_pred_benchmark
+    )
+
+    function_loss_with_weights = metric_func(
+        y_true,
+        y_pred,
+        y_train=y_train,
+        y_pred_benchmark=y_pred_benchmark,
+        sample_weight=weight_generator,
+    )
+
+    is_num = is_numeric_dtype(function_loss)
+    is_scalar = np.isscalar(function_loss)
+    assert is_num and is_scalar, " ".join(
+        ["Loss function with univariate input should return scalar number"]
+    )
+
+    is_num = is_numeric_dtype(function_loss_with_weights)
+    is_scalar = np.isscalar(function_loss_with_weights)
+    assert is_num and is_scalar, " ".join(
+        ["Loss function with sample weight generator should return scalar number"]
+    )
+
+    assert np.isclose(function_loss, function_loss_with_weights), " ".join(
+        ["Loss function with sample weight generator should return different value"]
+    )
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed(["sktime.performance_metrics"]),
+    reason="Run if performance_metrics module has changed.",
+)
+@pytest.mark.parametrize("random_state", RANDOM_STATES)
+@pytest.mark.parametrize(
+    "metric_class",
+    [
+        MeanSquaredError,
+        MeanAbsoluteError,
+        MeanAbsolutePercentageError,
+    ],
+)
+def test_sample_weight_generator_is_passed_to_func(metric_class, random_state):
+    """Tests that loss function with univariate input should return scalar number."""
+    y = _make_series(n_timepoints=75, random_state=random_state)
+    y_train, y_true = y.iloc[:50], y.iloc[50:]
+    y_pred = y.shift(1).iloc[50:]
+    y_pred_benchmark = y.rolling(2).mean().iloc[50:]
+
+    class TestWeightGenerator(BaseSampleWeightGenerator):
+        def __call__(self, y_true, y_pred=None, **kwargs):
+            return np.arange(1, len(y_true) + 1)
+
+    weight_generator = TestWeightGenerator()
+
+    metric = metric_class()
+
+    function_loss = metric(
+        y_true, y_pred, y_train=y_train, y_pred_benchmark=y_pred_benchmark
+    )
+    function_loss_with_weights = metric(
+        y_true,
+        y_pred,
+        y_train=y_train,
+        y_pred_benchmark=y_pred_benchmark,
+        sample_weight=weight_generator,
+    )
+
+    assert not np.isclose(function_loss, function_loss_with_weights), " ".join(
+        ["Loss function with sample weight generator should return different value"]
+    )