multitask.runner.run_with

multitask.runner.run_with(
    multitask_cls,
    all_tasks,
    unknown_task_message=None,
    config=None,
    task='lazy',
    dataframe=None,
    data_test=None,
    project_name='test_project',
    cache_home=None,
    plot_with_outliers=False,
    show=False,
    show_progress=False,
    dry_run=False,
    log_level=40,
    **overrides,
)

Parameterized core of the MultiTask pipeline orchestration.

This is the reusable seam that allows the sibling spotforecast2 package to bind an extended task class and a wider task set without duplicating the orchestration body. All caller-visible behaviour is identical to run().

Parameters

Name	Type	Description	Default
multitask_cls	type	The `MultiTask`-compatible class to instantiate. `spotforecast2-safe` passes its own `MultiTask`; the sibling package passes its subclass that adds tuning tasks.	required
all_tasks	FrozenSet[str]	The complete frozenset of permitted task names for this binding. `run()` passes `_ALL_TASKS`; an extended binding includes additional task names (e.g. `"optuna"`).	required
unknown_task_message	Optional[Callable[[str, List[str]], str]]	Optional callable `(task, sorted_tasks) -> str` that produces the `ValueError` message when `task` is not in `all_tasks`. When `None`, a default message mentioning that auto-tuning tasks require the `spotforecast2` sibling package is used.	`None`
config	Optional[PipelineConfig]	A `PipelineConfig`-conforming object (typically `ConfigMulti`). When `None`, a fresh `ConfigMulti()` is constructed with default fields.	`None`
task	str	Pipeline mode. Must be a member of `all_tasks`. Defaults to `"lazy"`.	`'lazy'`
dataframe	Optional[pd.DataFrame]	Input time-series data. Required for pipeline tasks; optional for `"clean"`.	`None`
data_test	Optional[pd.DataFrame]	Ground-truth DataFrame covering the prediction horizon. Optional.	`None`
project_name	str	Active-dataset identifier. Sets `config.data_frame_name`.	`'test_project'`
cache_home	Optional[str]	Cache directory override. When `None`, `get_cache_home()` is called to resolve the default.	`None`
plot_with_outliers	bool	When `True`, calls `mt.plot_with_outliers()` between detect_outliers and impute. Raises `NotImplementedError` in `spotforecast2-safe`.	`False`
show	bool	Forwarded to `mt.run(show=show)`.	`False`
show_progress	bool	Forwarded to `multitask_cls` constructor.	`False`
dry_run	bool	Forwarded to `multitask_cls` constructor.	`False`
log_level	int	Logging level. Defaults to 40 (ERROR).	`40`
**overrides	Any	Forwarded to the `multitask_cls` constructor, which applies them via `config.set_params(**overrides)` in `BaseTask.__init__`. Mutates the caller’s config object.	`{}`

Returns

Name	Type	Description
	pd.DataFrame	DataFrame whose index is the forecast horizon timestamps and
	pd.DataFrame	whose single column `"forecast"` contains the aggregated
	pd.DataFrame	predicted values. For the `"clean"` task an empty DataFrame is
	pd.DataFrame	returned.

Raises

Name	Type	Description
	ValueError	If `task` is not a member of `all_tasks`.

Examples

Run a lazy forecast via the low-level run_with seam, passing MultiTask and the safe-package task set directly:

import warnings
import tempfile
import pandas as pd
warnings.filterwarnings("ignore")

from spotforecast2_safe.multitask.runner import run_with, _ALL_TASKS
from spotforecast2_safe.multitask.multi import MultiTask
from spotforecast2_safe.configurator.config_multi import ConfigMulti
from spotforecast2_safe.data.fetch_data import fetch_data, get_package_data_home

data_home = get_package_data_home()
df = fetch_data(filename=str(data_home / "demo02.csv"))
cache_dir = tempfile.mkdtemp()

cfg = ConfigMulti(
    train_size=pd.Timedelta(days=30),
    predict_size=6,
    imputation_method="weighted",
    use_exogenous_features=False,
    auto_save_models=False,
    targets=["A"],
    number_folds=2,
    verbose=False,
)

forecast = run_with(
    multitask_cls=MultiTask,
    all_tasks=_ALL_TASKS,
    config=cfg,
    task="lazy",
    dataframe=df,
    project_name="demo02_runner",
    cache_home=cache_dir,
    show=False,
    show_progress=False,
    log_level=40,
)
print(forecast)
assert isinstance(forecast, pd.DataFrame)
assert "forecast" in forecast.columns
assert len(forecast) == 6

                           forecast
1975-06-18 19:00:00+00:00  0.073144
1975-06-18 20:00:00+00:00 -0.036825
1975-06-18 21:00:00+00:00 -0.005090
1975-06-18 22:00:00+00:00 -0.052408
1975-06-18 23:00:00+00:00 -0.026746
1975-06-19 00:00:00+00:00  0.002038

Passing an unsupported task name raises a ValueError with a descriptive message pointing to the sibling package:

from spotforecast2_safe.multitask.runner import run_with, _ALL_TASKS
from spotforecast2_safe.multitask.multi import MultiTask

try:
    run_with(
        multitask_cls=MultiTask,
        all_tasks=_ALL_TASKS,
        task="optuna",
    )
except ValueError as exc:
    print(exc)

Unknown task 'optuna'. Choose from: ['clean', 'defaults', 'lazy', 'predict']. Auto-tuning tasks ('optuna', 'spotoptim') require the spotforecast2 sibling package.