Repository: sktime/sktime
Pull request #5055 (2023-08-08 23:41:24)

[ENH] Extend benchmarking to accept sktime-in-memory data formats
#### Reference Issues/PRs
partly towards #4873 

#### What does this implement/fix? Explain your changes.
<!--
A clear and concise description of what you have implemented.
-->

PR to allow kotsu benchmarking to accept sktime-in-memory container (which is what is widely used across sktime as data ingestion input) listed in MTYPE_REGISTER (mostly from pd.DataFrame) for Series, Multivariate and Hierarchical time series. 

In addition, also allows users to access estimator predictions and training and test time points.

#### 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.
-->

I added  `test_dataset_different_format` to check for output consistency.

#### Any other comments?
<!--
Please be aware that we are a loose team of volunteers so patience is necessary; assistance handling other issues is very welcome. We value all user contributions, no matter how minor they are. If we are slow to review, either the pull request needs some benchmarking, tinkering, convincing, etc. or more likely the reviewers are simply busy. In either case, we ask for your understanding during the review process.
-->

There is a design question that I would like to discuss. The functions `forecasting_validation` and `_factory_forecasting_validation` are currently created as separate functions outside of the `ForecastingBenchmark` class. I propose that we can convert them into methods instead, to make `forecasting_validation` an abstract validation method, say, called `.prepare_forecasting_validation`.  By doing this, we allow users to substitute the evaluation procedure if they do not want to use the default `evaluate` method. Instead, they can use their own evaluation steps by subclassing `ForecastingBenchmark` and replacing the `prepare_forecasting_validation` method with their own evaluation steps.

are they any specific reasons why they were created as functions rather than method in the first place?

fyi @fkiraly @DBCerigo @alex-hh 

fkiraly: > are they any specific reasons why they were created as functions rather than method in the first place?

This is not my design - but I would agree that methods make more sense.

I also find the nested argument design where you pass functions to functions quite strange. Surely this can be solved more cleanly, e.g., with passing classes and dispatching?
hazrulakmal: ## SKTIME MTYPE

The enhancement made is to accept dataset_loader as a tuple of the format (y, X), where y and X are in sktime MTYPE. When X is not available, we expect dataset_loader as (y, None).

For a callable dataset_loader, previously it is assumed to return a single value, to accept a target variable. Hence, there's no way to allow X exogenous variable(s) in. This PR also makes an adjustment to account for a callable that returns two values, so that benchmarking can now accept exogenous variables as well when a callable is given.

Now to input X variable(s), users can do either of the following

```python
from sktime.datasets import load_longley
from sktime.benchmarking.forecasting import ForecastingBenchmark

y, X = load_longley() # an abstract pre-processed MTYPE of any datasets
benchmark = ForecastingBenchmark()
benchmark.add_task(
    (y, X),
    CV,
    scores
)

```

or

```python
from sktime.datasets import load_longley
from sktime.benchmarking.forecasting import ForecastingBenchmark

benchmark = ForecastingBenchmark()
benchmark.add_task(
    load_longley,
    CV,
    scores
)

```

## Estimator Predictions

Storing estimator predictions can be quite tricky. My initial solution is to simply pass the predictions, train and test sets from the `evaluate` output to the benchmark results dictionary. However, this will result in redundancy (in the order of estimators) of the train and test sets in the final output. this is a serious problem when the dataset is huge. 

Anyway, the reason for saving train and test sets on top of the prediction is to allow users to perform further post-processing analysis, such as statistical tests, visualizations etc, after running the benchmarks.

Currently, I am considering saving only the estimator predictions and finding another way to save the train and test sets for each fold.

Saving the estimator predictions can pose a problem when the resulting dataframe is saved as a CSV file. If a user wants to look at the benchmark result and loads the CSV file back into memory, the saved predictions may not load properly because they were saved as MTYPE (such as pd.dataframe). This is a general drawback of CSV files. One solution is to serialize the dataframe as a pickle, which can then be deserialized properly to the original form without issues that are present with CSV files. However, this also means that an override is needed to change the format in which the data is saved.
fkiraly: Regarding design questions, here are some ideas:

* I think the cleanest way would be to give `evaluate` a suitable signature, e.g., allowing to pass multiple estimators, and add the estimator name as another index. This would still multiplicate the data, but in a second step we could move/deprecate to a format where the true splits are returned in a separate output. Going with non-nested data containers would be better anyway.
* Now, a full change of `evaluate` might be too much for doing it in one step. In the interim, we could just discard anything that's multiplied, e.g., get data only from the first estimator?
* regarding storing the output: I think a clean way would be an option to pass a reference to an object, which is, if passed, where outputs are written. Or, more generally, a config such as `{"path": "path/to/file", "serializer": "pickle"}`. Going about it this way would allow to retain the status quo as a default until deprecation, and would allow more reasonable options like pickle, or using the object's own serializer (the `save` method!) 
hazrulakmal: > I think a clean way would be an option to pass a reference to an object

It seems that Kotsu has limitations when saving a nested container because it uses the default CSV saving option. Changing or extending this option would require a significant override of the Kotsu run function. Therefore, I have decided to deal with this issue separately in another PR to avoid cluttering this PR.


Patch:
diff --git a/sktime/benchmarking/forecasting.py b/sktime/benchmarking/forecasting.py
index 5dcc521e9..6f9719b9c 100644
--- a/sktime/benchmarking/forecasting.py
+++ b/sktime/benchmarking/forecasting.py
@@ -29,14 +29,27 @@ def forecasting_validation(
 
     Parameters
     ----------
-    dataset_loader : Callable
-        A function which returns a dataset, like from `sktime.datasets`.
+    dataset_loader : Callable or a tuple of sktime data objects, or sktime data object
+
+        * if sktime dataset object, must return a tuple of (y, X) upon calling ``load``,
+        where ``y`` is the endogenous data container and
+        ``X`` is the exogenous data container.
+        * If Callable, must be a function which returns an endogenous
+        data container ``y``, or a tuple with endogenous ``y`` and exogenous ``X``,
+        like from ``sktime.datasets``
+        * If Tuple, must be in the format of (y, X) where ``y`` is an endogenous
+        data container and ``X`` is an exogenous data container.
+        * if sktime data object, will be interpreted as endogenous ``y`` data container.
+
     cv_splitter : BaseSplitter object
         Splitter used for generating validation folds.
+
     scorers : a list of BaseMetric objects
         Each BaseMetric output will be included in the results.
+
     estimator : BaseForecaster object
         Estimator to benchmark.
+
     backend : {"dask", "loky", "multiprocessing", "threading"}, by default None.
         Runs parallel evaluate for each task if specified.
 
@@ -93,6 +106,7 @@ def forecasting_validation(
         Value to assign to the score if an exception occurs in estimator fitting. If set
         to "raise", the exception is raised. If a numeric value is given,
         FitFailedWarning is raised.
+
     strategy : {"refit", "update", "no-update_params"}, optional, default="refit"
         defines the ingestion mode when the forecaster sees new data when window expands
         "refit" = forecaster is refitted to each training window
@@ -103,36 +117,22 @@ def forecasting_validation(
     -------
     Dictionary of benchmark results for that forecaster
     """
-    y = dataset_loader()
+    data = _coerce_data_for_evaluate(dataset_loader)
 
-    results = {}
-    if isinstance(y, tuple):
-        y, X = y
-        scores_df = evaluate(
-            forecaster=estimator,
-            y=y,
-            X=X,
-            cv=cv_splitter,
-            scoring=scorers,
-            backend=backend,
-            backend_params=backend_params,
-            cv_global=cv_global,
-            error_score=error_score,
-            strategy=strategy,
-        )
-    else:
-        scores_df = evaluate(
-            forecaster=estimator,
-            y=y,
-            cv=cv_splitter,
-            scoring=scorers,
-            backend=backend,
-            backend_params=backend_params,
-            cv_global=cv_global,
-            error_score=error_score,
-            strategy=strategy,
-        )
+    scores_df = evaluate(
+        forecaster=estimator,
+        cv=cv_splitter,
+        scoring=scorers,
+        backend=backend,
+        backend_params=backend_params,
+        cv_global=cv_global,
+        error_score=error_score,
+        strategy=strategy,
+        **data,  # y and X
+    )
 
+    # collect results by scorer
+    results = {}
     for scorer in scorers:
         scorer_name = scorer.name
         for ix, row in scores_df.iterrows():
@@ -142,6 +142,24 @@ def forecasting_validation(
     return results
 
 
+def _coerce_data_for_evaluate(dataset_loader):
+    """Coerce data input object to a dict to pass to forecasting evaluate."""
+    if callable(dataset_loader) and not hasattr(dataset_loader, "load"):
+        data = dataset_loader()
+    elif callable(dataset_loader) and hasattr(dataset_loader, "load"):
+        data = dataset_loader.load()
+    else:
+        data = dataset_loader
+
+    if isinstance(data, tuple) and len(data) == 2:
+        y, X = data
+        return {"y": y, "X": X}
+    elif isinstance(data, tuple) and len(data) == 1:
+        return {"y": data[0]}
+    else:
+        return {"y": data}
+
+
 def _factory_forecasting_validation(
     dataset_loader: Callable,
     cv_splitter: BaseSplitter,
@@ -237,8 +255,11 @@ class ForecastingBenchmark(BaseBenchmark):
 
         Parameters
         ----------
-        dataset_loader : Callable
-            A function which returns a dataset, like from `sktime.datasets`.
+        dataset_loader : Callable or a tuple
+            If Callable. a function which returns a dataset, like from `sktime.datasets`
+            If Tuple, must be in the format of (Y, X) where Y is the target variable
+            and X is exogenous variabele where both must be sktime pd.DataFrame MTYPE.
+            When tuple is given, task_id argument must be filled.
         cv_splitter : BaseSplitter object
             Splitter used for generating validation folds.
         scorers : a list of BaseMetric objects


Test patch:
diff --git a/sktime/benchmarking/tests/test_forecasting.py b/sktime/benchmarking/tests/test_forecasting.py
index b167560e7..9ee22b6df 100644
--- a/sktime/benchmarking/tests/test_forecasting.py
+++ b/sktime/benchmarking/tests/test_forecasting.py
@@ -7,6 +7,7 @@ from sklearn.model_selection import KFold
 
 from sktime.benchmarking.benchmarks import coerce_estimator_and_id
 from sktime.benchmarking.forecasting import ForecastingBenchmark
+from sktime.datasets import load_airline, load_longley
 from sktime.forecasting.naive import NaiveForecaster
 from sktime.forecasting.trend import TrendForecaster
 from sktime.performance_metrics.forecasting import (
@@ -19,6 +20,8 @@ from sktime.tests.test_switch import run_test_module_changed
 from sktime.utils._testing.hierarchical import _make_hierarchical
 from sktime.utils.dependencies import _check_soft_dependencies
 
+# TODO:
+# Manual test is labor intensive, need to refactor the tests for fast iteration
 EXPECTED_RESULTS_1 = pd.DataFrame(
     data={
         "validation_id": "[dataset=data_loader_simple]_"
@@ -251,3 +254,31 @@ def test_multiple_estimators(estimators):
     assert len(registered_estimators) == len(
         estimators
     ), "add_estimator does not register all estimators."
+
+
+@pytest.mark.skipif(
+    not run_test_module_changed("sktime.benchmarking"),
+    reason="run test only if benchmarking module has changed",
+)
+def test_dataset_different_format(tmp_path):
+    """Test to check different dataset formats to output identical results."""
+    y_1, X_1 = load_longley()
+    y_2 = load_airline()
+    datasets = [[load_longley, load_airline], [(y_1, X_1), (y_2, None)]]
+    scorers = [MeanAbsolutePercentageError(), MeanAbsoluteError()]
+    cv_splitters = [
+        ExpandingWindowSplitter(initial_window=14),
+        ExpandingWindowSplitter(initial_window=142),
+    ]
+
+    outputs = []
+    benchmarks = [ForecastingBenchmark(), ForecastingBenchmark()]
+    for idx, benchmark in enumerate(benchmarks):
+        benchmark.add_estimator(NaiveForecaster(strategy="last"))
+        benchmark.add_task(datasets[idx][0], cv_splitters[0], scorers, f"{idx}-1")
+        benchmark.add_task(datasets[idx][1], cv_splitters[1], scorers, f"{idx}-2")
+        results_file = tmp_path / f"results_{idx}.csv"
+        output_df = benchmark.run(results_file)
+        outputs.append(output_df.drop(columns=["runtime_secs", "validation_id"]))
+
+    pd.testing.assert_frame_equal(outputs[0], outputs[1], check_exact=True)
diff --git a/sktime/tests/tests/test_test_utils.py b/sktime/tests/tests/test_test_utils.py
index cd5082dcb..fec1c5370 100644
--- a/sktime/tests/tests/test_test_utils.py
+++ b/sktime/tests/tests/test_test_utils.py
@@ -29,12 +29,13 @@ def test_excluded_tests_by_test():
         )
     ]
     excluded_estimators = EXCLUDED_TESTS_BY_TEST["test_get_test_params_coverage"]
-    assert set(excluded_estimators) - set(EXCLUDE_SOFT_DEPS) == set(
+    assert set(excluded_estimators) - set(EXCLUDE_SOFT_DEPS) <= set(
         filtered_estimators
     ) - set(EXCLUDE_SOFT_DEPS), (
-        "Assertion failed: The sets of excluded and filtered estimators "
-        "do not match. Please remove the estimator you have added "
-        "test parameters to from EXCLUDED_TESTS_BY_TEST or EXCLUDE_SOFT_DEPS."
+        "If this PR adds test parameters to an estimator's get_test_params: "
+        "Please ensure to remove this estimator "
+        "from EXCLUDED_TESTS_BY_TEST and EXCLUDE_SOFT_DEPS "
+        "in sktime.tests._config, if it is present there."
     )