Dataset Submodule

SRToolkit.dataset

Datasets and benchmarks for symbolic regression.

A dataset (SR_dataset) represents a single equation with associated data, a symbol library, and evaluation constraints. A benchmark (SR_benchmark) is a collection of datasets.

Modules:

Name	Description
sr_dataset	SR_dataset — wraps input data and evaluation settings for a single equation discovery problem.
sr_benchmark	SR_benchmark — manages a collection of datasets.
feynman	Feynman — 100-equation physics benchmark.
nguyen	Nguyen — 10-equation polynomial/trig benchmark.
srsd_feynman	SRSD_Feynman — 120-equation SRSD physics benchmark with per-variable sampling strategies.
sampling	LogUniformSampling, UniformSampling, IntegerUniformSampling — variable samplers with serialisation support.
data_source	DataSource and its concrete types UrlSource and SampleSource — describe where a dataset's cached data originates.
data_cache	dataset cache management — list, gc, remove, refresh, dataset_path — plus the materialisation engine used internally by the dataset machinery.

DataSource

Bases: ABC

Abstract base class describing the origin of a dataset's cached data.

Concrete subclasses must implement to_dict, from_dict, and materialize. The dict produced by to_dict must include a "source_class" key holding the fully-qualified class path (e.g. "SRToolkit.dataset.data_source.UrlSource") so that DataSource.from_dict can reconstruct it via importlib without a central registry.

The cache layer stores a hash of every source's data_source + samplers config and warns when it drifts from what the cache was built with — regardless of this flag — so a changed url, changed n_samples/seed, or a switch between source kinds is always surfaced.

Attributes:

Name	Type	Description
is_volatile	bool	Informational hint: True when the materialised output depends on parameters that can change between runs (e.g. sampler ranges or seed), as opposed to a fixed external artifact (a pinned url). Defaults to False. Does not affect drift detection, which runs for all sources.

to_dict abstractmethod

to_dict() -> dict

Serialize this source to a JSON-compatible dictionary.

The returned dict must include "source_class" set to the fully-qualified class path of this source.

Source code in SRToolkit/dataset/data_source.py

@abstractmethod
def to_dict(self) -> dict:
    """
    Serialize this source to a JSON-compatible dictionary.

    The returned dict **must** include ``"source_class"`` set to the fully-qualified
    class path of this source.
    """

from_dict classmethod

from_dict(d: Optional[dict]) -> Optional['DataSource']

Reconstruct a source from a dict produced by to_dict.

When called on the base DataSource class, dispatches to the concrete subclass named by the "source_class" key using importlib — both built-in and user-defined subclasses round-trip without a central registry. When called on a concrete subclass, that subclass must override this method.

Parameters:

Name	Type	Description	Default
d	Optional[dict]	Dictionary with a "source_class" key (fully-qualified class path) and the source's parameters, or None (data is already cached / supplied directly).	required

Returns:

Type	Description
Optional['DataSource']	A reconstructed DataSource instance,
Optional['DataSource']	or None if d is None.

Raises:

Type	Description
KeyError	If "source_class" is missing from d (dispatch path).
ImportError	If the class cannot be imported (dispatch path).
NotImplementedError	If called on a subclass that has not overridden this method.

Source code in SRToolkit/dataset/data_source.py

@classmethod
def from_dict(cls, d: Optional[dict]) -> Optional["DataSource"]:
    """
    Reconstruct a source from a dict produced by
    [to_dict][SRToolkit.dataset.data_source.DataSource.to_dict].

    When called on the base [DataSource][SRToolkit.dataset.data_source.DataSource]
    class, dispatches to the concrete subclass named by the ``"source_class"`` key
    using ``importlib`` — both built-in and user-defined subclasses round-trip without
    a central registry. When called on a concrete subclass, that subclass must override
    this method.

    Args:
        d: Dictionary with a ``"source_class"`` key (fully-qualified class path) and the
            source's parameters, or ``None`` (data is already cached / supplied directly).

    Returns:
        A reconstructed [DataSource][SRToolkit.dataset.data_source.DataSource] instance,
        or ``None`` if ``d`` is ``None``.

    Raises:
        KeyError: If ``"source_class"`` is missing from ``d`` (dispatch path).
        ImportError: If the class cannot be imported (dispatch path).
        NotImplementedError: If called on a subclass that has not overridden this method.
    """
    if cls is DataSource:
        if d is None:
            return None
        d = _auto_bind(d)
        class_path = d["source_class"]
        module_path, cls_name = class_path.rsplit(".", 1)
        try:
            resolved = getattr(importlib.import_module(module_path), cls_name)
        except (ImportError, AttributeError):
            raise ImportError(
                f"Cannot import data source class {class_path!r}. "
                "If this is a bundle class, install the bundle first. "
                "If the config has no '_bundle' key, call bind_config(config) manually."
            ) from None
        return resolved.from_dict(d)
    raise NotImplementedError(f"{cls.__name__}.from_dict is not implemented.")

materialize abstractmethod

materialize(cache_path: Path, config: dict) -> None

Produce the dataset's data at cache_path (an .npz file with X and, for RMSE datasets, y).

Parameters:

Name	Type	Description	Default
cache_path	Path	Target .npz path inside the data cache.	required
config	dict	The full serialised dataset config (see SR_dataset.to_dict). Sources that generate data (e.g. SampleSource) read the dataset's "samplers", "ground_truth", "symbol_library", and "ranking_function" from it.	required

Source code in SRToolkit/dataset/data_source.py

@abstractmethod
def materialize(self, cache_path: Path, config: dict) -> None:
    """
    Produce the dataset's data at ``cache_path`` (an ``.npz`` file with ``X`` and,
    for RMSE datasets, ``y``).

    Args:
        cache_path: Target ``.npz`` path inside the data cache.
        config: The full serialised dataset config (see
            [SR_dataset.to_dict][SRToolkit.dataset.sr_dataset.SR_dataset.to_dict]).
            Sources that generate data (e.g.
            [SampleSource][SRToolkit.dataset.data_source.SampleSource]) read the
            dataset's ``"samplers"``, ``"ground_truth"``, ``"symbol_library"``, and
            ``"ranking_function"`` from it.
    """

FallbackSource

FallbackSource(sources: List[DataSource])

Bases: DataSource

A chain of data sources tried in order until one succeeds.

materialize attempts each source's materialize in turn; if one raises, a warning is emitted and the next is tried. The first to succeed wins; if all fail, a RuntimeError chaining the last error is raised.

The built-in benchmarks use this to prefer canonical downloaded data while keeping a network-free fallback — FallbackSource([UrlSource(archive), SampleSource(seed=...)]) downloads the authoritative archive when reachable and regenerates from the dataset's samplers otherwise. Because the whole chain is serialised in the config, the preference travels: a grid recipe, an export, or a cold worker each download the canonical data first and only sample if it is unavailable — unlike a bare SampleSource, which would always regenerate.

Parameters:

Name	Type	Description	Default
sources	List[DataSource]	Ordered, non-empty list of DataSource to try.	required

Source code in SRToolkit/dataset/data_source.py

def __init__(self, sources: List[DataSource]):
    if not sources:
        raise ValueError("[FallbackSource] requires at least one source.")
    self.sources = list(sources)
    # ``True`` if any source in the chain is volatile (e.g. a ``SampleSource``).
    self.is_volatile = any(s.is_volatile for s in self.sources)

to_dict

to_dict() -> dict

Serialize this source and its whole chain to a JSON-compatible dictionary.

Source code in SRToolkit/dataset/data_source.py

def to_dict(self) -> dict:
    """Serialize this source and its whole chain to a JSON-compatible dictionary."""
    return {
        "source_class": "SRToolkit.dataset.data_source.FallbackSource",
        "sources": [s.to_dict() for s in self.sources],
    }

from_dict classmethod

from_dict(d: Optional[dict]) -> 'FallbackSource'

Deserialize a FallbackSource, reconstructing each child source.

Source code in SRToolkit/dataset/data_source.py

@classmethod
def from_dict(cls, d: Optional[dict]) -> "FallbackSource":
    """Deserialize a [FallbackSource][SRToolkit.dataset.data_source.FallbackSource], reconstructing each child source."""
    if d is None:
        raise ValueError("[FallbackSource.from_dict] requires a dictionary, got None.")
    sources = []
    for s in d["sources"]:
        ds = DataSource.from_dict(s)
        if ds is not None:
            sources.append(ds)
    if len(sources) > 0:
        return cls(sources)
    else:
        raise ValueError(
            f"[FallbackSource.from_dict] Could not construct at least one DataSource for dictionary: {json.dumps(d)}"
        )

materialize

materialize(cache_path: Path, config: dict) -> None

Try each source in order; warn and continue on failure, re-raising if all fail.

Source code in SRToolkit/dataset/data_source.py

def materialize(self, cache_path: Path, config: dict) -> None:
    """Try each source in order; warn and continue on failure, re-raising if all fail."""
    last_exc: Optional[Exception] = None
    for i, source in enumerate(self.sources):
        try:
            source.materialize(cache_path, config)
            return
        except Exception as exc:  # noqa: BLE001 - fall through to the next source in the chain
            last_exc = exc
            if i + 1 < len(self.sources):
                warnings.warn(
                    f"[FallbackSource] {type(source).__name__} failed to materialise "
                    f"({exc}); trying the next fallback.",
                    stacklevel=2,
                )
    raise RuntimeError(
        f"[FallbackSource] All {len(self.sources)} sources failed to materialise; last error: {last_exc}"
    ) from last_exc

SampleSource

SampleSource(n_samples: int = 10000, seed: Optional[int] = None)

Bases: DataSource

Data generated by drawing n_samples points from the dataset's samplers.

This source does not own the samplers — it carries only the generation parameters and reads the samplers from the dataset config at materialisation time. The dataset must therefore define samplers (one per input variable). For RMSE datasets with a token-list ground truth, the targets y are produced by evaluating that expression on the generated inputs.

Because the output depends on the samplers and seed, this source is volatile: the cache layer records a hash and warns when the configuration drifts (call refresh to regenerate).

Parameters:

Name	Type	Description	Default
n_samples	int	Number of input rows to generate. Defaults to 10000.	10000
seed	Optional[int]	Random seed for reproducible generation. None means no seed is set.	None

Source code in SRToolkit/dataset/data_source.py

def __init__(self, n_samples: int = 10000, seed: Optional[int] = None):
    self.n_samples = n_samples
    self.seed = seed

to_dict

to_dict() -> dict

Serialize this source to a JSON-compatible dictionary.

Source code in SRToolkit/dataset/data_source.py

def to_dict(self) -> dict:
    """Serialize this source to a JSON-compatible dictionary."""
    return {
        "source_class": "SRToolkit.dataset.data_source.SampleSource",
        "n_samples": self.n_samples,
        "seed": self.seed,
    }

from_dict classmethod

from_dict(d: Optional[dict]) -> 'SampleSource'

Deserialize a SampleSource from a dictionary.

Source code in SRToolkit/dataset/data_source.py

@classmethod
def from_dict(cls, d: Optional[dict]) -> "SampleSource":
    """Deserialize a [SampleSource][SRToolkit.dataset.data_source.SampleSource] from a dictionary."""
    if d is None:
        raise ValueError("[SampleSource.from_dict] requires a dictionary, got None.")
    return cls(n_samples=d.get("n_samples", 10000), seed=d.get("seed"))

materialize

materialize(cache_path: Path, config: dict) -> None

Generate X (and y for RMSE) from the dataset's samplers and save them.

Source code in SRToolkit/dataset/data_source.py

def materialize(self, cache_path: Path, config: dict) -> None:
    """Generate ``X`` (and ``y`` for RMSE) from the dataset's samplers and save them."""
    samplers_raw = config.get("samplers")
    if not samplers_raw:
        raise ValueError(
            "[SampleSource.materialize] Cannot generate data: the dataset defines no "
            "'samplers'. A SampleSource requires samplers (one per input variable)."
        )

    if self.seed is not None:
        np.random.seed(self.seed)

    samplers = [Sampler.from_dict(s) for s in samplers_raw]
    X = np.column_stack([s(self.n_samples) for s in samplers])

    ranking_function = config.get("ranking_function", "rmse")
    ground_truth = config.get("ground_truth")
    y = None

    if ranking_function == "rmse":
        if ground_truth is None:
            raise ValueError(
                "[SampleSource.materialize] Cannot generate data: ranking_function is "
                "'rmse' but the dataset has no 'ground_truth'. A SampleSource produces "
                "targets 'y' by evaluating the ground-truth expression on the sampled "
                "inputs, so a ground truth is required. Provide one, or supply X/y "
                "directly (data_source=None)."
            )
        if isinstance(ground_truth, np.ndarray):
            raise ValueError(
                "[SampleSource.materialize] ranking_function is 'rmse' but 'ground_truth' "
                "is a numpy array (a behaviour matrix). Behaviour matrices are a 'bed' "
                "concept and cannot be evaluated to produce targets 'y'. Provide the "
                "ground truth as a token list or Node expression for RMSE datasets."
            )
        sl_dict = config.get("symbol_library")
        if sl_dict is not None:
            sl = SymbolLibrary.from_dict(sl_dict)
        else:
            sl = SymbolLibrary.default_symbols(X.shape[1])
        f = compile_expr(ground_truth, sl)
        y = f(X, np.array([]))

    cache_path.parent.mkdir(parents=True, exist_ok=True)
    if y is not None:
        np.savez(str(cache_path), X=X, y=y)
    else:
        np.savez(str(cache_path), X=X)

UrlSource

UrlSource(url: str)

Bases: DataSource

Data downloaded as a .zip archive from a URL and extracted into the cache.

Two archive layouts are accepted transparently: a flat zip whose <dataset_name>.npz files sit at the root (as served by the built-in benchmarks), or a to_archive archive whose data lives under a data/ prefix (in which case the prefix is stripped and the bundled benchmark.json / dataset.json is ignored). Either way, the expected <dataset_name>.npz must end up in the version directory.

Parameters:

Name	Type	Description	Default
url	str	URL of a .zip archive.	required

Source code in SRToolkit/dataset/data_source.py

def __init__(self, url: str):
    self.url = url

to_dict

to_dict() -> dict

Serialize this source to a JSON-compatible dictionary.

Source code in SRToolkit/dataset/data_source.py

def to_dict(self) -> dict:
    """Serialize this source to a JSON-compatible dictionary."""
    return {"source_class": "SRToolkit.dataset.data_source.UrlSource", "url": self.url}

from_dict classmethod

from_dict(d: Optional[dict]) -> 'UrlSource'

Deserialize a UrlSource from a dictionary.

Source code in SRToolkit/dataset/data_source.py

@classmethod
def from_dict(cls, d: Optional[dict]) -> "UrlSource":
    """Deserialize a [UrlSource][SRToolkit.dataset.data_source.UrlSource] from a dictionary."""
    if d is None:
        raise ValueError("[UrlSource.from_dict] requires a dictionary, got None.")
    return cls(d["url"])

materialize

materialize(cache_path: Path, config: dict) -> None

Download the archive and extract its data files into the version directory.

Source code in SRToolkit/dataset/data_source.py

def materialize(self, cache_path: Path, config: dict) -> None:
    """Download the archive and extract its data files into the version directory."""
    from SRToolkit.dataset import data_cache

    version_dir = cache_path.parent
    version_dir.mkdir(parents=True, exist_ok=True)

    http_response = urlopen(self.url)
    with ZipFile(BytesIO(http_response.read())) as zf:
        data_cache.extract_zip_into_version_dir(zf, version_dir)

    if not cache_path.exists():
        raise RuntimeError(
            f"[UrlSource.materialize] After downloading from '{self.url}', the expected "
            f"file '{cache_path}' still does not exist."
        )

Feynman

Feynman(n_samples: int = 10000, seed: Optional[int] = 42, force_generate: bool = False)

Bases: SR_benchmark

The Feynman symbolic regression benchmark.

Contains 100 physics equations with up to 9 variables. Data is downloaded on first use from the SymbolicRegressionToolkit repository (10,000 samples per dataset instead of the original 1,000,000 from the paper). If the download fails, data is generated from the stored per-variable samplers using n_samples points and the given seed.

References

Udrescu & Tegmark (2020)

Examples:

>>> benchmark = Feynman()
>>> len(benchmark.list_datasets(verbose=False))
100

Parameters:

Name	Type	Description	Default
n_samples	int	Number of samples to generate per dataset when force_generate=True (sampler-based data generation). Defaults to 10000.	10000
seed	Optional[int]	Random seed used for sampler-based data generation. Defaults to 42.	42
force_generate	bool	If True, generate fresh data from the stored samplers instead of downloading the pre-generated data. Defaults to False.	False

Source code in SRToolkit/dataset/feynman.py

def __init__(
    self,
    n_samples: int = 10000,
    seed: Optional[int] = 42,
    force_generate: bool = False,
):
    super().__init__("feynman", version="1.0.0")
    self._n_samples = n_samples
    self._seed = seed
    self._force_generate = force_generate
    self._populate()

Nguyen

Nguyen(n_samples: int = 10000, seed: Optional[int] = 42, force_generate: bool = False)

Bases: SR_benchmark

The Nguyen symbolic regression benchmark.

Contains 10 expressions without constant parameters (first 4 are polynomials, first 8 use one variable, last 2 use two variables). The benchmark ships with pre-generated data. If the download fails, data is generated from the stored per-variable samplers using n_samples points and the given seed.

References

Uy et al. (2011)

Examples:

>>> benchmark = Nguyen()
>>> len(benchmark.list_datasets(verbose=False))
10

Parameters:

Name	Type	Description	Default
n_samples	int	Number of samples to generate per dataset when force_generate=True (sampler-based data generation). Defaults to 10000.	10000
seed	Optional[int]	Random seed used for sampler-based data generation. Defaults to 42.	42
force_generate	bool	If True, generate fresh data from the stored samplers instead of downloading the pre-generated data. Defaults to False.	False

Source code in SRToolkit/dataset/nguyen.py

def __init__(
    self,
    n_samples: int = 10000,
    seed: Optional[int] = 42,
    force_generate: bool = False,
):
    super().__init__("Nguyen", version="1.0.0")
    self._n_samples = n_samples
    self._seed = seed
    self._force_generate = force_generate
    self._populate()

IntegerUniformSampling

IntegerUniformSampling(min_value: int, max_value: int, uses_positive: bool = True, uses_negative: bool = True)

Bases: Sampler

Integer uniform sampler with configurable sign constraints.

Samples integers from :math:\{\text{min}, ..., \text{max}-1\}, optionally drawing from positive and/or negative ranges.

Parameters:

Name	Type	Description	Default
min_value	int	Lower bound of the integer range.	required
max_value	int	Upper bound (exclusive) of the integer range.	required
uses_positive	bool	If True, positive samples are included.	True
uses_negative	bool	If True, negative samples are included.	True

Source code in SRToolkit/dataset/sampling.py

def __init__(self, min_value: int, max_value: int, uses_positive: bool = True, uses_negative: bool = True):
    self.min_value = int(min_value)
    self.max_value = int(max_value)
    assert uses_positive or uses_negative
    self.uses_positive = uses_positive
    self.uses_negative = uses_negative

to_dict

to_dict() -> dict

Serialize this sampler to a JSON-compatible dictionary.

Source code in SRToolkit/dataset/sampling.py

def to_dict(self) -> dict:
    """Serialize this sampler to a JSON-compatible dictionary."""
    return {
        "sampler_class": "SRToolkit.dataset.sampling.IntegerUniformSampling",
        "min_value": self.min_value,
        "max_value": self.max_value,
        "uses_positive": self.uses_positive,
        "uses_negative": self.uses_negative,
    }

from_dict classmethod

from_dict(d: dict) -> IntegerUniformSampling

Deserialize a IntegerUniformSampling from a dictionary produced by to_dict.

Source code in SRToolkit/dataset/sampling.py

@classmethod
def from_dict(cls, d: dict) -> "IntegerUniformSampling":
    """Deserialize a [IntegerUniformSampling][SRToolkit.dataset.sampling.IntegerUniformSampling] from a dictionary produced by [to_dict][SRToolkit.dataset.sampling.IntegerUniformSampling.to_dict]."""
    return cls(d["min_value"], d["max_value"], d["uses_positive"], d["uses_negative"])

LogUniformSampling

LogUniformSampling(min_value: float, max_value: float, uses_positive: bool = True, uses_negative: bool = True)

Bases: Sampler

Log-uniform sampler with configurable sign constraints.

Samples from U(\log_{10}(\text{min}), \log_{10}(\text{max})) in log space, optionally drawing from positive and/or negative ranges.

Parameters:

Name	Type	Description	Default
min_value	float	Lower bound of the log-uniform range (must be > 0).	required
max_value	float	Upper bound of the log-uniform range (must be > 0).	required
uses_positive	bool	If True, positive samples are included.	True
uses_negative	bool	If True, negative samples are included.	True

Source code in SRToolkit/dataset/sampling.py

def __init__(self, min_value: float, max_value: float, uses_positive: bool = True, uses_negative: bool = True):
    self.min_value = min_value
    self.max_value = max_value
    assert uses_positive or uses_negative
    self.uses_positive = uses_positive
    self.uses_negative = uses_negative

to_dict

to_dict() -> dict

Serialize this sampler to a JSON-compatible dictionary.

Source code in SRToolkit/dataset/sampling.py

def to_dict(self) -> dict:
    """Serialize this sampler to a JSON-compatible dictionary."""
    return {
        "sampler_class": "SRToolkit.dataset.sampling.LogUniformSampling",
        "min_value": self.min_value,
        "max_value": self.max_value,
        "uses_positive": self.uses_positive,
        "uses_negative": self.uses_negative,
    }

from_dict classmethod

from_dict(d: dict) -> LogUniformSampling

Deserialize a LogUniformSampling from a dictionary produced by to_dict.

Source code in SRToolkit/dataset/sampling.py

@classmethod
def from_dict(cls, d: dict) -> "LogUniformSampling":
    """Deserialize a [LogUniformSampling][SRToolkit.dataset.sampling.LogUniformSampling] from a dictionary produced by [to_dict][SRToolkit.dataset.sampling.LogUniformSampling.to_dict]."""
    return cls(d["min_value"], d["max_value"], d["uses_positive"], d["uses_negative"])

Sampler

Bases: ABC

Abstract base class for variable samplers.

Concrete subclasses must implement __call__, to_dict, and from_dict. The dictionary produced by to_dict must include a "sampler_class" key holding the fully-qualified class path (e.g. "SRToolkit.dataset.sampling.UniformSampling"), so that Sampler.from_dict can reconstruct any subclass — including user-defined ones — via importlib without a central registry.

__call__ abstractmethod

__call__(sample_size: int) -> np.ndarray

Draw sample_size samples and return them as a 1-D numpy array.

Source code in SRToolkit/dataset/sampling.py

@abstractmethod
def __call__(self, sample_size: int) -> np.ndarray:
    """Draw ``sample_size`` samples and return them as a 1-D numpy array."""

to_dict abstractmethod

to_dict() -> dict

Serialize this sampler to a JSON-compatible dictionary.

The returned dict must include "sampler_class" set to the fully-qualified class path of this sampler.

Source code in SRToolkit/dataset/sampling.py

@abstractmethod
def to_dict(self) -> dict:
    """
    Serialize this sampler to a JSON-compatible dictionary.

    The returned dict **must** include ``"sampler_class"`` set to the
    fully-qualified class path of this sampler.
    """

from_dict classmethod

from_dict(d: dict) -> Sampler

Reconstruct a sampler from a dictionary produced by to_dict.

When called on the base Sampler class, dispatches to the concrete subclass named by the "sampler_class" key using importlib — both built-in and user-defined subclasses round-trip without a central registry. When called on a concrete subclass, that subclass must override this method.

Parameters:

Name	Type	Description	Default
d	dict	Dictionary with a "sampler_class" key (fully-qualified class path, e.g. "SRToolkit.dataset.sampling.UniformSampling") and the sampler's parameters.	required

Returns:

Type	Description
Sampler	A reconstructed Sampler instance.

Raises:

Type	Description
KeyError	If "sampler_class" is missing from d (dispatch path).
ImportError	If the class cannot be imported (dispatch path).
NotImplementedError	If called on a subclass that has not overridden this method.

Source code in SRToolkit/dataset/sampling.py

@classmethod
def from_dict(cls, d: dict) -> "Sampler":
    """
    Reconstruct a sampler from a dictionary produced by
    [to_dict][SRToolkit.dataset.sampling.Sampler.to_dict].

    When called on the base [Sampler][SRToolkit.dataset.sampling.Sampler] class,
    dispatches to the concrete subclass named by the ``"sampler_class"`` key using
    ``importlib`` — both built-in and user-defined subclasses round-trip without a
    central registry. When called on a concrete subclass, that subclass must override
    this method.

    Args:
        d: Dictionary with a ``"sampler_class"`` key (fully-qualified class path, e.g.
            ``"SRToolkit.dataset.sampling.UniformSampling"``) and the sampler's parameters.

    Returns:
        A reconstructed [Sampler][SRToolkit.dataset.sampling.Sampler] instance.

    Raises:
        KeyError: If ``"sampler_class"`` is missing from ``d`` (dispatch path).
        ImportError: If the class cannot be imported (dispatch path).
        NotImplementedError: If called on a subclass that has not overridden this method.
    """
    if cls is Sampler:
        d = _auto_bind(d)
        class_path = d["sampler_class"]
        module_path, cls_name = class_path.rsplit(".", 1)
        try:
            resolved = getattr(importlib.import_module(module_path), cls_name)
        except (ImportError, AttributeError):
            raise ImportError(
                f"Cannot import sampler class {class_path!r}. "
                "If this is a bundle class, install the bundle first. "
                "If the config has no '_bundle' key, call bind_config(config) manually."
            ) from None
        return resolved.from_dict(d)
    raise NotImplementedError(f"{cls.__name__}.from_dict is not implemented.")

UniformSampling

UniformSampling(min_value: float, max_value: float, uses_positive: bool = True, uses_negative: bool = True)

Bases: Sampler

Linear uniform sampler with configurable sign constraints.

Samples fromU(\text{min}, \text{max}), optionally drawing from positive and/or negative ranges.

Parameters:

Name	Type	Description	Default
min_value	float	Lower bound of the uniform range.	required
max_value	float	Upper bound of the uniform range.	required
uses_positive	bool	If True, positive samples are included.	True
uses_negative	bool	If True, negative samples are included.	True

Source code in SRToolkit/dataset/sampling.py

def __init__(self, min_value: float, max_value: float, uses_positive: bool = True, uses_negative: bool = True):
    self.min_value = min_value
    self.max_value = max_value
    assert uses_positive or uses_negative
    self.uses_positive = uses_positive
    self.uses_negative = uses_negative

to_dict

to_dict() -> dict

Serialize this sampler to a JSON-compatible dictionary.

Source code in SRToolkit/dataset/sampling.py

def to_dict(self) -> dict:
    """Serialize this sampler to a JSON-compatible dictionary."""
    return {
        "sampler_class": "SRToolkit.dataset.sampling.UniformSampling",
        "min_value": self.min_value,
        "max_value": self.max_value,
        "uses_positive": self.uses_positive,
        "uses_negative": self.uses_negative,
    }

from_dict classmethod

from_dict(d: dict) -> UniformSampling

Deserialize a UniformSampling from a dictionary produced by to_dict.

Source code in SRToolkit/dataset/sampling.py

@classmethod
def from_dict(cls, d: dict) -> "UniformSampling":
    """Deserialize a [UniformSampling][SRToolkit.dataset.sampling.UniformSampling] from a dictionary produced by [to_dict][SRToolkit.dataset.sampling.UniformSampling.to_dict]."""
    return cls(d["min_value"], d["max_value"], d["uses_positive"], d["uses_negative"])

SR_benchmark

SR_benchmark(benchmark_name: str, datasets: Optional[List[Union[SR_dataset, Tuple[str, SR_dataset]]]] = None, metadata: Optional[Dict[str, Any]] = None, version: str = '1.0.0', base_dir: Optional[str] = None)

A named, persistent collection of symbolic regression datasets.

Examples:

>>> from SRToolkit.dataset import Feynman
>>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
>>> len(benchmark.list_datasets(verbose=False))
100

Parameters:

Name	Type	Description	Default
benchmark_name	str	Name of this benchmark.	required
datasets	Optional[List[Union[SR_dataset, Tuple[str, SR_dataset]]]]	Initial datasets to add. Each element can be an SR_dataset instance (auto-named as "<benchmark_name>_<index>") or a (name, SR_dataset) tuple.	None
metadata	Optional[Dict[str, Any]]	Optional dictionary of benchmark-level metadata (e.g. citation, description).	None
version	str	Version string for this benchmark. Defaults to "1.0.0".	'1.0.0'
base_dir	Optional[str]	Directory where dataset files are stored or will be written. Optional — if omitted, the data cache is used exclusively.	None

Raises:

Type	Description
ValueError	If any element of datasets is not an SR_dataset or a valid (name, SR_dataset) tuple.

Source code in SRToolkit/dataset/sr_benchmark.py

def __init__(
    self,
    benchmark_name: str,
    datasets: Optional[List[Union[SR_dataset, Tuple[str, SR_dataset]]]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    version: str = "1.0.0",
    base_dir: Optional[str] = None,
):
    """
    A named, persistent collection of symbolic regression datasets.

    Examples:
        >>> from SRToolkit.dataset import Feynman
        >>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
        >>> len(benchmark.list_datasets(verbose=False))
        100

    Args:
        benchmark_name: Name of this benchmark.
        datasets: Initial datasets to add. Each element can be an
            [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] instance (auto-named as
            ``"<benchmark_name>_<index>"``) or a ``(name, SR_dataset)`` tuple.
        metadata: Optional dictionary of benchmark-level metadata (e.g. citation, description).
        version: Version string for this benchmark. Defaults to ``"1.0.0"``.
        base_dir: Directory where dataset files are stored or will be written.
            Optional — if omitted, the data cache is used exclusively.

    Raises:
        ValueError: If any element of ``datasets`` is not an
            [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] or a valid ``(name, SR_dataset)`` tuple.
    """
    self.benchmark_name = benchmark_name
    self.base_dir = base_dir
    self.version = version
    self.datasets: Dict[str, Dict[str, Any]] = {}
    self.metadata = {} if metadata is None else metadata
    if datasets is not None:
        for i, dataset in enumerate(datasets):
            if isinstance(dataset, SR_dataset):
                self.add_dataset_instance(benchmark_name + "_" + str(i + 1), dataset)
            elif isinstance(dataset, tuple) and isinstance(dataset[0], str) and isinstance(dataset[1], SR_dataset):
                self.add_dataset_instance(dataset[0], dataset[1])
            else:
                raise ValueError(
                    "[SR_benchmark] Dataset inside the datasets argument must be either a tuple "
                    "(name, SR_dataset) or a SR_dataset instance."
                )

add_dataset_instance

add_dataset_instance(dataset_name: str, dataset: SR_dataset)

Adds an instance of the SR_dataset class to the benchmark.

Examples:

>>> from SRToolkit.dataset import Feynman
>>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
>>> dataset = benchmark.create_dataset('I.16.6')
>>> isinstance(dataset, SR_dataset)
True
>>> bm = SR_benchmark("BM")
>>> bm.add_dataset_instance("I.16.6", dataset)

Parameters:

Name	Type	Description	Default
dataset_name	str	The name of the dataset.	required
dataset	SR_dataset	An instance of the SR_dataset class.	required

Raises:

Type	Description
Exception	If the dataset name already exists in the benchmark.

Source code in SRToolkit/dataset/sr_benchmark.py

def add_dataset_instance(self, dataset_name: str, dataset: SR_dataset):
    """
    Adds an instance of the SR_dataset class to the benchmark.

    Examples:
        >>> from SRToolkit.dataset import Feynman
        >>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
        >>> dataset = benchmark.create_dataset('I.16.6')
        >>> isinstance(dataset, SR_dataset)
        True
        >>> bm = SR_benchmark("BM")
        >>> bm.add_dataset_instance("I.16.6", dataset)

    Args:
         dataset_name: The name of the dataset.
         dataset: An instance of the SR_dataset class.

    Raises:
        Exception: If the dataset name already exists in the benchmark.
    """
    if dataset_name in self.datasets:
        raise ValueError(f"Dataset {dataset_name} already exists in the benchmark.")
    else:
        self.datasets[dataset_name] = {}
    self.datasets[dataset_name]["sr_dataset"] = dataset
    self.datasets[dataset_name]["num_variables"] = dataset.X.shape[1]

add_dataset

add_dataset(symbol_library: SymbolLibrary, dataset: Optional[Union[ndarray, Tuple[ndarray, ndarray]]] = None, dataset_name: Optional[str] = None, ranking_function: str = 'rmse', max_evaluations: int = -1, ground_truth: Optional[Union[List[str], Node, ndarray]] = None, original_equation: Optional[str] = None, success_threshold: Optional[float] = None, seed: Optional[int] = None, dataset_metadata: Optional[dict] = None, samplers: Optional[List[Sampler]] = None, data_source: Optional[DataSource] = None, **kwargs: Unpack[EstimationSettings])

Adds a dataset to the benchmark.

Examples:

>>> import tempfile, numpy as np
>>> from SRToolkit.utils import SymbolLibrary
>>> bm = SR_benchmark("BM")
>>> X = np.random.rand(10, 2)
>>> y = X[:, 0] + X[:, 1]
>>> bm.add_dataset(SymbolLibrary.default_symbols(2),(X, y),dataset_name="test_ds",ground_truth=["X_0", "+", "X_1"],original_equation="y = x0 + x1")
>>> len(bm.list_datasets(verbose=False))
1

Parameters:

Name	Type	Description	Default
dataset	Optional[Union[ndarray, Tuple[ndarray, ndarray]]]	Direct data for the dataset: a 2-D numpy array (features) or a (X, y) tuple. Use None together with data_source to have the cache layer materialise the data instead. When data_source is provided this argument is ignored. To use data from a local file, load it yourself (e.g. with numpy.load) and pass the arrays here.	None
symbol_library	SymbolLibrary	The symbol library to use.	required
dataset_name	Optional[str]	The name of the dataset. Auto-generated if None.	None
ranking_function	str	"rmse" or "bed".	'rmse'
max_evaluations	int	Maximum expressions to evaluate. -1 means no limit.	-1
ground_truth	Optional[Union[List[str], Node, ndarray]]	Ground truth expression.	None
original_equation	Optional[str]	Human-readable equation string.	None
success_threshold	Optional[float]	Error threshold for success. None means no threshold.	None
seed	Optional[int]	Random seed.	None
dataset_metadata	Optional[dict]	Optional dataset-level metadata dict.	None
samplers	Optional[List[Sampler]]	Optional list of samplers (one per input variable). They define the problem's input distribution and power resample; a SampleSource draws from them.	None
data_source	Optional[DataSource]	Optional DataSource describing where the data comes from (UrlSource or SampleSource). When provided, the dataset argument is ignored and the cache layer manages materialisation.	None
**kwargs	Unpack[EstimationSettings]	Estimation settings forwarded to SR_evaluator.	{}

Raises:

Type	Description
ValueError	Various validation errors (see below).

Source code in SRToolkit/dataset/sr_benchmark.py

def add_dataset(
    self,
    symbol_library: SymbolLibrary,
    dataset: Optional[Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]] = None,
    dataset_name: Optional[str] = None,
    ranking_function: str = "rmse",
    max_evaluations: int = -1,
    ground_truth: Optional[Union[List[str], Node, np.ndarray]] = None,
    original_equation: Optional[str] = None,
    success_threshold: Optional[float] = None,
    seed: Optional[int] = None,
    dataset_metadata: Optional[dict] = None,
    samplers: Optional[List[Sampler]] = None,
    data_source: Optional[DataSource] = None,
    **kwargs: Unpack[EstimationSettings],
):
    """
    Adds a dataset to the benchmark.

    Examples:
        >>> import tempfile, numpy as np
        >>> from SRToolkit.utils import SymbolLibrary
        >>> bm = SR_benchmark("BM")
        >>> X = np.random.rand(10, 2)
        >>> y = X[:, 0] + X[:, 1]
        >>> bm.add_dataset(SymbolLibrary.default_symbols(2),(X, y),dataset_name="test_ds",ground_truth=["X_0", "+", "X_1"],original_equation="y = x0 + x1")
        >>> len(bm.list_datasets(verbose=False))
        1

    Args:
        dataset: Direct data for the dataset: a 2-D numpy array (features) or a
            ``(X, y)`` tuple. Use ``None`` together with ``data_source`` to have the
            cache layer materialise the data instead. When ``data_source`` is provided
            this argument is ignored. To use data from a local file, load it
            yourself (e.g. with ``numpy.load``) and pass the arrays here.
        symbol_library: The symbol library to use.
        dataset_name: The name of the dataset. Auto-generated if ``None``.
        ranking_function: ``"rmse"`` or ``"bed"``.
        max_evaluations: Maximum expressions to evaluate. ``-1`` means no limit.
        ground_truth: Ground truth expression.
        original_equation: Human-readable equation string.
        success_threshold: Error threshold for success. ``None`` means no threshold.
        seed: Random seed.
        dataset_metadata: Optional dataset-level metadata dict.
        samplers: Optional list of samplers (one per input variable). They define the
            problem's input distribution and power
            [resample][SRToolkit.dataset.sr_dataset.SR_dataset.resample]; a
            [SampleSource][SRToolkit.dataset.data_source.SampleSource] draws from them.
        data_source: Optional [DataSource][SRToolkit.dataset.data_source.DataSource]
            describing where the data comes from
            ([UrlSource][SRToolkit.dataset.data_source.UrlSource] or
            [SampleSource][SRToolkit.dataset.data_source.SampleSource]). When provided,
            the ``dataset`` argument is ignored and the cache layer manages
            materialisation.
        **kwargs: Estimation settings forwarded to
            [SR_evaluator][SRToolkit.evaluation.sr_evaluator.SR_evaluator].

    Raises:
        ValueError: Various validation errors (see below).
    """

    if dataset_name is None:
        dataset_name = f"{self.benchmark_name}_{len(self.datasets) + 1}"

    # Fail fast before any cache files are written, mirroring add_dataset_instance.
    if dataset_name in self.datasets:
        raise ValueError(f"[SR_benchmark.add_dataset] Dataset '{dataset_name}' already exists in the benchmark.")

    if "bed_X" in kwargs and kwargs["bed_X"] is not None:
        kwargs["bed_X"] = kwargs["bed_X"].tolist()

    # An ndarray ('bed' behaviour matrix) ground truth is not JSON-safe: the entry
    # stores None and the array is written to <name>_gt.npy at the end.
    if isinstance(ground_truth, Node):
        ground_truth_out: Optional[Union[List[str], str]] = ground_truth.to_list()
    elif isinstance(ground_truth, np.ndarray):
        ground_truth_out = None
    else:
        ground_truth_out = ground_truth

    merged_metadata = copy.deepcopy(self.metadata)
    if dataset_metadata:
        merged_metadata.update(dataset_metadata)

    entry: Dict[str, Any] = {
        "format_version": 2,
        "dataset_name": dataset_name,
        "benchmark": self.benchmark_name,
        "version": self.version,
        "symbol_library": symbol_library.to_dict(),
        "ranking_function": ranking_function,
        "max_evaluations": max_evaluations,
        "success_threshold": success_threshold,
        "seed": seed,
        "dataset_metadata": merged_metadata,
        "original_equation": original_equation,
        "kwargs": kwargs,
        "ground_truth": ground_truth_out,
        "samplers": [s.to_dict() for s in samplers] if samplers is not None else None,
        # Default count: one variable per sampler, else the symbol library. The array
        # branches below override this with the real column count when data is supplied.
        "num_variables": len(samplers) if samplers is not None else _count_variables(symbol_library),
    }

    # Derive a human-readable equation from a token-list ground truth if absent.
    if ground_truth is None:
        if ranking_function == "bed":
            raise ValueError("[SR_benchmark.add_dataset] For 'bed' ranking, the ground truth must be provided.")
        warnings.warn(
            "[SR_benchmark.add_dataset] 'ground_truth' argument not provided. We recommend providing it "
            "for more transparent evaluation."
        )
    elif original_equation is None and isinstance(ground_truth, list):
        entry["original_equation"] = "y = " + "".join(ground_truth)

    # Resolve the data source / write the arrays. num_variables defaults to the sampler /
    # symbol-library count set above; the array branches override it with the real shape.
    if data_source is not None:
        # Lazy: the cache layer materialises X (and y) on first use; nothing written here.
        entry["data_source"] = data_source.to_dict()
    else:
        entry["data_source"] = None
        if isinstance(dataset, np.ndarray):
            if ranking_function == "rmse":
                if ground_truth is None:
                    raise ValueError(
                        "[SR_benchmark.add_dataset] For 'rmse' ranking, if the dataset argument is a numpy "
                        "array, the ground truth must be provided."
                    )
                if isinstance(ground_truth, np.ndarray):
                    raise ValueError(
                        "[SR_benchmark.add_dataset] For 'rmse' ranking, the ground truth must be "
                        "a list of tokens from the symbol library or a SRToolkit.utils.Node object."
                    )
                try:
                    y = compile_expr(ground_truth, symbol_library)(dataset, None)
                except Exception as e:
                    raise Exception(
                        f"[SR_benchmark.add_dataset] Could not evaluate the ground truth. Original error: {e}"
                    )
                _save_arrays_to_cache(self.benchmark_name, self.version, dataset_name, dataset, y)
            elif ranking_function == "bed":
                _save_arrays_to_cache(self.benchmark_name, self.version, dataset_name, dataset)

        elif isinstance(dataset, tuple):
            if (
                len(dataset) != 2
                or not isinstance(dataset[0], np.ndarray)
                or not isinstance(dataset[1], np.ndarray)
            ):
                raise ValueError(
                    "[SR_benchmark.add_dataset] When the dataset argument is a tuple, it must be (X, y) "
                    "with both values numpy arrays."
                )
            if ranking_function == "bed":
                warnings.warn(
                    "[SR_benchmark.add_dataset] 'bed' ranking only utilizes the array with features. "
                    "Array with targets will be ignored."
                )
            _save_arrays_to_cache(self.benchmark_name, self.version, dataset_name, dataset[0], dataset[1])

    if isinstance(ground_truth, np.ndarray):
        _save_gt_array_to_cache(self.benchmark_name, self.version, dataset_name, ground_truth)

    self.datasets[dataset_name] = entry

add_from_samplers

add_from_samplers(ground_truth: Union[List[str], Node], samplers: List[Sampler], symbol_library: Optional[SymbolLibrary] = None, n_samples: int = 10000, seed: Optional[int] = None, ranking_function: str = 'rmse', dataset_name: Optional[str] = None, original_equation: Optional[str] = None, success_threshold: Optional[float] = None, max_evaluations: int = -1, dataset_metadata: Optional[dict] = None, **kwargs: Unpack[EstimationSettings]) -> None

Add a dataset described only by a ground-truth expression and per-variable samplers.

This is the benchmark-level counterpart to SR_dataset.from_samplers: it attaches a SampleSource so the data is generated lazily from samplers (and, for RMSE, ground_truth) the first time the dataset is materialised via create_dataset.

Examples:

>>> from SRToolkit.dataset.sampling import UniformSampling
>>> bm = SR_benchmark("BM")
>>> bm.add_from_samplers(["X_0", "+", "X_1"],
...     [UniformSampling(0, 5), UniformSampling(0, 5)], dataset_name="add",
...     n_samples=100, seed=0)
>>> ds = bm.create_dataset("add")
>>> ds.X.shape
(100, 2)

Parameters:

Name	Type	Description	Default
ground_truth	Union[List[str], Node]	Ground-truth expression as a token list or Node.	required
samplers	List[Sampler]	One Sampler per input variable.	required
symbol_library	Optional[SymbolLibrary]	Token vocabulary. Defaults to default_symbols with one variable per sampler.	None
n_samples	int	Number of input rows to generate on materialisation. Defaults to 10000.	10000
seed	Optional[int]	Random seed stored on the SampleSource.	None
ranking_function	str	"rmse" or "bed".	'rmse'
dataset_name	Optional[str]	Name of the dataset. Auto-generated if None.	None
original_equation	Optional[str]	Human-readable equation string. Auto-filled from a token-list ground_truth when None.	None
success_threshold	Optional[float]	Error threshold for success. None means no threshold.	None
max_evaluations	int	Maximum expressions to evaluate. -1 means no limit.	-1
dataset_metadata	Optional[dict]	Optional dataset-level metadata dict.	None
**kwargs	Unpack[EstimationSettings]	Estimation settings forwarded to SR_evaluator.	{}

Source code in SRToolkit/dataset/sr_benchmark.py

def add_from_samplers(
    self,
    ground_truth: Union[List[str], Node],
    samplers: List[Sampler],
    symbol_library: Optional[SymbolLibrary] = None,
    n_samples: int = 10000,
    seed: Optional[int] = None,
    ranking_function: str = "rmse",
    dataset_name: Optional[str] = None,
    original_equation: Optional[str] = None,
    success_threshold: Optional[float] = None,
    max_evaluations: int = -1,
    dataset_metadata: Optional[dict] = None,
    **kwargs: Unpack[EstimationSettings],
) -> None:
    """
    Add a dataset described only by a ground-truth expression and per-variable samplers.

    This is the benchmark-level counterpart to
    [SR_dataset.from_samplers][SRToolkit.dataset.sr_dataset.SR_dataset.from_samplers]:
    it attaches a [SampleSource][SRToolkit.dataset.data_source.SampleSource] so the data
    is generated lazily from ``samplers`` (and, for RMSE, ``ground_truth``) the first
    time the dataset is materialised via
    [create_dataset][SRToolkit.dataset.sr_benchmark.SR_benchmark.create_dataset].

    Examples:
        >>> from SRToolkit.dataset.sampling import UniformSampling
        >>> bm = SR_benchmark("BM")
        >>> bm.add_from_samplers(["X_0", "+", "X_1"],
        ...     [UniformSampling(0, 5), UniformSampling(0, 5)], dataset_name="add",
        ...     n_samples=100, seed=0)
        >>> ds = bm.create_dataset("add")
        >>> ds.X.shape
        (100, 2)

    Args:
        ground_truth: Ground-truth expression as a token list or
            [Node][SRToolkit.utils.expression_tree.Node].
        samplers: One [Sampler][SRToolkit.dataset.sampling.Sampler] per input variable.
        symbol_library: Token vocabulary. Defaults to
            [default_symbols][SRToolkit.utils.symbol_library.SymbolLibrary.default_symbols]
            with one variable per sampler.
        n_samples: Number of input rows to generate on materialisation. Defaults to ``10000``.
        seed: Random seed stored on the
            [SampleSource][SRToolkit.dataset.data_source.SampleSource].
        ranking_function: ``"rmse"`` or ``"bed"``.
        dataset_name: Name of the dataset. Auto-generated if ``None``.
        original_equation: Human-readable equation string. Auto-filled from a token-list
            ``ground_truth`` when ``None``.
        success_threshold: Error threshold for success. ``None`` means no threshold.
        max_evaluations: Maximum expressions to evaluate. ``-1`` means no limit.
        dataset_metadata: Optional dataset-level metadata dict.
        **kwargs: Estimation settings forwarded to
            [SR_evaluator][SRToolkit.evaluation.sr_evaluator.SR_evaluator].
    """
    if symbol_library is None:
        if not samplers:
            raise ValueError(
                "[SR_benchmark.add_from_samplers] 'samplers' must be a non-empty list "
                "(one sampler per input variable)."
            )
        symbol_library = SymbolLibrary.default_symbols(len(samplers))

    self.add_dataset(
        symbol_library=symbol_library,
        dataset=None,
        dataset_name=dataset_name,
        ranking_function=ranking_function,
        max_evaluations=max_evaluations,
        ground_truth=ground_truth,
        original_equation=original_equation,
        success_threshold=success_threshold,
        seed=seed,
        dataset_metadata=dataset_metadata,
        samplers=samplers,
        data_source=SampleSource(n_samples=n_samples, seed=seed),
        **kwargs,
    )

create_dataset

create_dataset(dataset_name: str, n_samples: Optional[int] = None, seed: Optional[int] = None) -> SR_dataset

Creates an instance of a dataset from the given dataset name.

When n_samples is provided the returned dataset contains freshly sampled data instead of the pre-generated data on disk. The dataset must have samplers defined (see samplers argument of add_dataset).

Examples:

>>> from SRToolkit.dataset import Feynman
>>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
>>> dataset = benchmark.create_dataset('I.16.6')
>>> dataset.X.shape
(10000, 3)
>>> dataset_small = benchmark.create_dataset('I.16.6', n_samples=500, seed=0)
>>> dataset_small.X.shape
(500, 3)

Parameters:

Name	Type	Description	Default
dataset_name	str	The name of the dataset to create.	required
n_samples	Optional[int]	If provided, generate a fresh dataset with this many samples using the stored samplers instead of loading pre-generated data from disk.	None
seed	Optional[int]	Random seed used when n_samples is provided. If None, no seed is set.	None

Returns:

Type	Description
SR_dataset	An SR_dataset instance containing the
SR_dataset	data, ground truth expression, and metadata for the given dataset.

Raises:

Type	Description
ValueError	If the dataset name is not found, or if n_samples is provided but the dataset has no samplers defined.

Source code in SRToolkit/dataset/sr_benchmark.py

def create_dataset(
    self,
    dataset_name: str,
    n_samples: Optional[int] = None,
    seed: Optional[int] = None,
) -> SR_dataset:
    """
    Creates an instance of a dataset from the given dataset name.

    When ``n_samples`` is provided the returned dataset contains freshly sampled data
    instead of the pre-generated data on disk. The dataset must have samplers defined
    (see ``samplers`` argument of
    [add_dataset][SRToolkit.dataset.sr_benchmark.SR_benchmark.add_dataset]).

    Examples:
        >>> from SRToolkit.dataset import Feynman
        >>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
        >>> dataset = benchmark.create_dataset('I.16.6')
        >>> dataset.X.shape
        (10000, 3)
        >>> dataset_small = benchmark.create_dataset('I.16.6', n_samples=500, seed=0)
        >>> dataset_small.X.shape
        (500, 3)

    Args:
        dataset_name: The name of the dataset to create.
        n_samples: If provided, generate a fresh dataset with this many samples using
            the stored samplers instead of loading pre-generated data from disk.
        seed: Random seed used when ``n_samples`` is provided. If ``None``, no seed is set.

    Returns:
        An [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] instance containing the
        data, ground truth expression, and metadata for the given dataset.

    Raises:
        ValueError: If the dataset name is not found, or if ``n_samples`` is provided but
            the dataset has no samplers defined.
    """
    if dataset_name not in self.datasets:
        raise ValueError(f"Dataset {dataset_name} not found")

    if n_samples is not None:
        config = self.datasets[dataset_name]
        if config.get("samplers") is None:
            raise ValueError(
                f"[SR_benchmark.create_dataset] Cannot resample '{dataset_name}': no samplers defined."
            )
        # _create_from_entry may return the benchmark's stored SR_dataset instance
        # (for entries added via add_dataset_instance). Copy before mutating X/y so
        # resampling never corrupts the stored dataset or aliases across callers.
        dataset = copy.deepcopy(self._create_from_entry(config))
        return dataset.resample_inplace(n_samples, seed=seed)

    # Loading canonical data: materialisation is driven entirely by each dataset's
    # ``data_source`` — the built-ins use a FallbackSource that downloads the canonical
    # archive once and regenerates from samplers only if the download is unavailable.
    return self._create_from_entry(self.datasets[dataset_name])

list_datasets

list_datasets(verbose: bool = True, num_variables: int = -1) -> List[str]

Lists the available datasets.

Examples:

>>> from SRToolkit.dataset import Feynman
>>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
>>> len(benchmark.list_datasets(num_variables=2, verbose=False))
15
>>> datasets_with_8_vars = benchmark.list_datasets(num_variables=8, verbose=False)
>>> datasets_with_8_vars[0]
'II.36.38'

Parameters:

Name	Type	Description	Default
verbose	bool	If True, also prints a description of each dataset.	True
num_variables	int	If not -1, only return datasets with this many input variables.	-1

Returns:

Type	Description
List[str]	A list of dataset names.

Source code in SRToolkit/dataset/sr_benchmark.py

def list_datasets(self, verbose: bool = True, num_variables: int = -1) -> List[str]:
    """
    Lists the available datasets.

    Examples:
        >>> from SRToolkit.dataset import Feynman
        >>> benchmark = Feynman() # Feynman is a specific instance of SR_benchmark with additional functionality
        >>> len(benchmark.list_datasets(num_variables=2, verbose=False))
        15
        >>> datasets_with_8_vars = benchmark.list_datasets(num_variables=8, verbose=False)
        >>> datasets_with_8_vars[0]
        'II.36.38'

    Args:
        verbose: If ``True``, also prints a description of each dataset.
        num_variables: If not ``-1``, only return datasets with this many input variables.

    Returns:
        A list of dataset names.
    """
    datasets = [
        dataset_name
        for dataset_name in self.datasets
        if num_variables < 0 or self.datasets[dataset_name].get("num_variables") == num_variables
    ]
    datasets = sorted(
        datasets,
        key=lambda dataset_name: (
            self.datasets[dataset_name].get("num_variables", -1),
            dataset_name,
        ),
    )

    if verbose:
        part1 = []
        part2 = []
        part3 = []
        max_length_1 = 0
        max_length_2 = 0
        for d in datasets:
            nv = self.datasets[d].get("num_variables", -1)
            if nv == 1:
                variable_str = "1 variable"
            elif nv is None or nv < 1:
                variable_str = "Amount of variables unknown"
            else:
                variable_str = f"{nv} variables"
            part1.append(d + ":")
            part2.append(variable_str)
            part3.append(self.datasets[d].get("original_equation"))
            if len(d) + 1 > max_length_1:
                max_length_1 = len(d) + 1
            if len(variable_str) > max_length_2:
                max_length_2 = len(variable_str)

        for p1, p2, p3 in zip(part1, part2, part3):
            print(f"{p1:<{max_length_1}} {p2:<{max_length_2}}, Expression: {p3}")
    return datasets

to_dict

to_dict() -> dict

Serialise the benchmark to a pure JSON-safe dictionary.

Dataset entries that have an sr_dataset key (added via add_dataset_instance) are serialised via SR_dataset.to_dict().

Returns:

Type	Description
dict	A JSON-safe dict representing the full benchmark configuration.

Source code in SRToolkit/dataset/sr_benchmark.py

def to_dict(self) -> dict:
    """
    Serialise the benchmark to a pure JSON-safe dictionary.

    Dataset entries that have an ``sr_dataset`` key (added via
    [add_dataset_instance][SRToolkit.dataset.sr_benchmark.SR_benchmark.add_dataset_instance])
    are serialised via ``SR_dataset.to_dict()``.

    Returns:
        A JSON-safe dict representing the full benchmark configuration.
    """
    datasets_out = {}
    for name, entry in self.datasets.items():
        if "sr_dataset" in entry:
            datasets_out[name] = entry["sr_dataset"].to_dict()
        else:
            datasets_out[name] = {k: v for k, v in entry.items() if k != "sr_dataset"}

    return {
        "format_version": 2,
        "type": "SR_benchmark",
        "benchmark_name": self.benchmark_name,
        "version": self.version,
        "metadata": self.metadata,
        "datasets": datasets_out,
    }

from_dict classmethod

from_dict(d: Union[dict, str, Path]) -> SR_benchmark

Reconstruct an SR_benchmark from a config dict or a saved JSON file.

To load a self-contained .zip archive (written by to_archive) use from_archive instead.

Parameters:

Name	Type	Description	Default
d	Union[dict, str, Path]	A dict produced by to_dict, or a path to a JSON file.	required

Returns:

Type	Description
SR_benchmark	An SR_benchmark instance.

Source code in SRToolkit/dataset/sr_benchmark.py

@classmethod
def from_dict(cls, d: Union[dict, str, Path]) -> "SR_benchmark":
    """
    Reconstruct an SR_benchmark from a config dict or a saved JSON file.

    To load a self-contained ``.zip`` archive (written by
    [to_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_archive]) use
    [from_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.from_archive] instead.

    Args:
        d: A dict produced by [to_dict][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_dict],
            or a path to a JSON file.

    Returns:
        An [SR_benchmark][SRToolkit.dataset.sr_benchmark.SR_benchmark] instance.
    """
    if isinstance(d, (str, Path)):
        if str(d).endswith(".zip"):
            raise ValueError(
                "[SR_benchmark.from_dict] Received a '.zip' path. Load self-contained "
                "archives with SR_benchmark.from_archive(path) instead."
            )
        with open(d) as f:
            dd = json.load(f)
    else:
        dd = d

    benchmark_name = dd["benchmark_name"]
    version = dd.get("version", "1.0.0")
    metadata = dd.get("metadata", {})

    b = cls(benchmark_name, version=version, metadata=metadata)

    for name, entry in dd.get("datasets", {}).items():
        # Store the entry dict directly; materialisation is lazy via create_dataset
        b.datasets[name] = entry
        if "num_variables" not in entry:
            # Try to infer
            samplers = entry.get("samplers")
            if samplers is not None:
                entry["num_variables"] = len(samplers)
            else:
                entry["num_variables"] = -1

    return b

to_archive

to_archive(path: Union[str, Path]) -> None

Write the benchmark (config + data) to a .zip archive.

The archive contains:

• benchmark.json: the benchmark configuration dict.
• data/<dataset_name>.npz: the cached data for each dataset.
• data/<dataset_name>_gt.npy: ground-truth behaviour array (if present).

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Destination path for the archive. Non-.zip suffixes trigger a warning but are still accepted.	required

Source code in SRToolkit/dataset/sr_benchmark.py

def to_archive(self, path: Union[str, Path]) -> None:
    """
    Write the benchmark (config + data) to a ``.zip`` archive.

    The archive contains:

    - ``benchmark.json``: the benchmark configuration dict.
    - ``data/<dataset_name>.npz``: the cached data for each dataset.
    - ``data/<dataset_name>_gt.npy``: ground-truth behaviour array (if present).

    Args:
        path: Destination path for the archive.  Non-``.zip`` suffixes trigger a
            warning but are still accepted.
    """
    path = Path(path)
    if path.suffix.lower() != ".zip":
        warnings.warn(
            f"[SR_benchmark.to_archive] path '{path}' does not end in '.zip'. "
            "The file will be a ZIP archive regardless of the extension.",
            stacklevel=2,
        )

    benchmark_json = json.dumps(self.to_dict(), indent=2)

    with zipfile.ZipFile(str(path), "w", compression=zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("benchmark.json", benchmark_json)

        for name, entry in self.datasets.items():
            if "sr_dataset" in entry:
                # In-memory dataset — write its data
                ds = entry["sr_dataset"]
                cache_p = data_cache.dataset_path(self.benchmark_name, self.version, name)
                if not cache_p.exists():
                    # Write directly from arrays
                    buf = io.BytesIO()
                    if ds.y is not None:
                        np.savez(buf, X=ds.X, y=ds.y)
                    else:
                        np.savez(buf, X=ds.X)
                    zf.writestr(f"data/{name}.npz", buf.getvalue())
                    if isinstance(ds.ground_truth, np.ndarray):
                        gt_buf = io.BytesIO()
                        np.save(gt_buf, ds.ground_truth)
                        zf.writestr(f"data/{name}_gt.npy", gt_buf.getvalue())
                    continue
            else:
                ds_name = entry.get("dataset_name", name)
                try:
                    cache_p = data_cache.resolve(self.benchmark_name, self.version, ds_name, entry)
                except Exception as e:
                    warnings.warn(f"[SR_benchmark.to_archive] Could not materialise '{name}': {e}. Skipping.")
                    continue

            zf.write(str(cache_p), f"data/{name}.npz")

            gt_path = cache_p.parent / f"{name}_gt.npy"
            if gt_path.exists():
                zf.write(str(gt_path), f"data/{name}_gt.npy")

from_archive classmethod

from_archive(path: Union[str, Path]) -> SR_benchmark

Load a benchmark from a self-contained .zip archive.

This is the counterpart to to_archive: it reads benchmark.json from the archive, extracts the bundled data/*.npz (and any _gt.npy) into the data cache, and returns a benchmark whose datasets read from that populated cache.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to a .zip archive written by to_archive.	required

Returns:

Type	Description
SR_benchmark	An SR_benchmark instance.

Source code in SRToolkit/dataset/sr_benchmark.py

@classmethod
def from_archive(cls, path: Union[str, Path]) -> "SR_benchmark":
    """
    Load a benchmark from a self-contained ``.zip`` archive.

    This is the counterpart to
    [to_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_archive]: it reads
    ``benchmark.json`` from the archive, extracts the bundled ``data/*.npz`` (and any
    ``_gt.npy``) into the data cache, and returns a benchmark whose datasets read
    from that populated cache.

    Args:
        path: Path to a ``.zip`` archive written by
            [to_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_archive].

    Returns:
        An [SR_benchmark][SRToolkit.dataset.sr_benchmark.SR_benchmark] instance.
    """
    with zipfile.ZipFile(str(path), "r") as zf:
        benchmark_dict = json.loads(zf.read("benchmark.json"))

    benchmark_name = benchmark_dict["benchmark_name"]
    version = benchmark_dict.get("version", "1.0.0")

    # Extract data into the cache
    data_cache.import_archive(Path(path), benchmark_name, version)

    b = cls(benchmark_name, version=version, metadata=benchmark_dict.get("metadata", {}))
    for name, entry in benchmark_dict.get("datasets", {}).items():
        b.datasets[name] = entry
        if "num_variables" not in entry:
            entry["num_variables"] = -1

    return b

from_url classmethod

from_url(url: str) -> SR_benchmark

Download a self-contained .zip archive from a URL and load it.

This is the remote counterpart to from_archive: the archive is downloaded to a temporary file and then loaded exactly as from_archive would. The url must point at an archive written by to_archive (a benchmark.json plus a data/ directory) — not a bare .npz/data zip (that is what UrlSource is for).

Parameters:

Name	Type	Description	Default
url	str	URL of a .zip archive written by to_archive.	required

Returns:

Type	Description
SR_benchmark	An SR_benchmark instance.

Source code in SRToolkit/dataset/sr_benchmark.py

@classmethod
def from_url(cls, url: str) -> "SR_benchmark":
    """
    Download a self-contained ``.zip`` archive from a URL and load it.

    This is the remote counterpart to
    [from_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.from_archive]: the
    archive is downloaded to a temporary file and then loaded exactly as
    ``from_archive`` would. The ``url`` must point at an archive written by
    [to_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_archive] (a
    ``benchmark.json`` plus a ``data/`` directory) — not a bare ``.npz``/data zip
    (that is what [UrlSource][SRToolkit.dataset.data_source.UrlSource] is for).

    Args:
        url: URL of a ``.zip`` archive written by
            [to_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_archive].

    Returns:
        An [SR_benchmark][SRToolkit.dataset.sr_benchmark.SR_benchmark] instance.
    """
    with urlopen(url) as response:
        data = response.read()

    tmp = tempfile.NamedTemporaryFile(suffix=".zip", delete=False)
    try:
        tmp.write(data)
        tmp.close()
        return cls.from_archive(tmp.name)
    finally:
        os.unlink(tmp.name)

SR_dataset

SR_dataset(X: ndarray, symbol_library: SymbolLibrary, ranking_function: str = 'rmse', y: Optional[ndarray] = None, max_evaluations: int = -1, ground_truth: Optional[Union[List[str], Node, ndarray]] = None, original_equation: Optional[str] = None, success_threshold: Optional[float] = None, seed: Optional[int] = None, dataset_metadata: Optional[dict] = None, dataset_name: str = 'unnamed', samplers: Optional[List[Sampler]] = None, benchmark: Optional[str] = None, version: Optional[str] = None, **kwargs: Unpack[EstimationSettings])

Wraps input data and evaluation settings for a single symbolic regression problem.

Examples:

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y", success_threshold=1e-6)
>>> evaluator = dataset.create_evaluator()
>>> bool(evaluator.evaluate_expr(["sin", "(", "X_0", ")"]) < dataset.success_threshold)
False
>>> bool(evaluator.evaluate_expr(["u-", "C", "*", "X_1", "+", "X_0"]) < dataset.success_threshold)
True

Parameters:

Name	Type	Description	Default
X	ndarray	Input data of shape (num_samples, num_variables) used to evaluate expressions.	required
symbol_library	SymbolLibrary	The symbol library defining the tokens used for the discovery task.	required
ranking_function	str	Ranking function to use. "rmse" calculates the RMSE between ground truth values and expression outputs with fitted free parameters. "bed" is a stochastic measure of behavioral distance between expressions; it is less sensitive to overfitting and focuses more on structure identification (see bed for more details).	'rmse'
y	Optional[ndarray]	Target values used for parameter estimation when ranking_function="rmse".	None
max_evaluations	int	Maximum number of expressions to evaluate. Values less than 0 mean no limit.	-1
ground_truth	Optional[Union[List[str], Node, ndarray]]	The ground truth expression, as a list of tokens in infix notation, a Node tree, or a numpy array of behavior matrix (see create_behavior_matrix). Numpy array is only applicable when ranking_function="bed".	None
original_equation	Optional[str]	Human-readable string of the original equation (e.g. "z = x + y").	None
success_threshold	Optional[float]	Error threshold below which an expression is considered successful. If None, no threshold is applied.	None
seed	Optional[int]	Random seed for reproducibility. None means no seed is set.	None
dataset_metadata	Optional[dict]	Optional dictionary of metadata about the dataset (e.g. citation, variable names).	None
dataset_name	str	Name for this dataset. Defaults to "unnamed".	'unnamed'
samplers	Optional[List[Sampler]]	Optional list of Sampler instances (one per input variable). The built-in LogUniformSampling, UniformSampling, and IntegerUniformSampling implement this interface.	None
benchmark	Optional[str]	Optional benchmark name (e.g. "feynman"). Required for serialisation via to_dict.	None
version	Optional[str]	Optional version string (e.g. "1.0.0"). Required for serialisation.	None
**kwargs	Unpack[EstimationSettings]	Optional estimation settings passed to SR_evaluator. Supported keys: method, tol, gtol, max_iter, constant_bounds, initialization, max_constants, max_expr_length, num_points_sampled, bed_X, num_consts_sampled, domain_bounds.	{}

Source code in SRToolkit/dataset/sr_dataset.py

def __init__(
    self,
    X: np.ndarray,
    symbol_library: SymbolLibrary,
    ranking_function: str = "rmse",
    y: Optional[np.ndarray] = None,
    max_evaluations: int = -1,
    ground_truth: Optional[Union[List[str], Node, np.ndarray]] = None,
    original_equation: Optional[str] = None,
    success_threshold: Optional[float] = None,
    seed: Optional[int] = None,
    dataset_metadata: Optional[dict] = None,
    dataset_name: str = "unnamed",
    samplers: Optional[List[Sampler]] = None,
    benchmark: Optional[str] = None,
    version: Optional[str] = None,
    **kwargs: Unpack[EstimationSettings],
) -> None:
    """
    Wraps input data and evaluation settings for a single symbolic regression problem.

    Examples:
        >>> X = np.array([[1, 2], [3, 4], [5, 6]])
        >>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
        ...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y", success_threshold=1e-6)
        >>> evaluator = dataset.create_evaluator()
        >>> bool(evaluator.evaluate_expr(["sin", "(", "X_0", ")"]) < dataset.success_threshold)
        False
        >>> bool(evaluator.evaluate_expr(["u-", "C", "*", "X_1", "+", "X_0"]) < dataset.success_threshold)
        True

    Args:
        X: Input data of shape ``(num_samples, num_variables)`` used to evaluate expressions.
        symbol_library: The symbol library defining the tokens used for the discovery task.
        ranking_function: Ranking function to use. ``"rmse"`` calculates the RMSE between ground truth
            values and expression outputs with fitted free parameters. ``"bed"`` is a stochastic measure of
            behavioral distance between expressions; it is less sensitive to overfitting and focuses more on
            structure identification (see [bed][SRToolkit.utils.measures.bed] for more details).
        y: Target values used for parameter estimation when ``ranking_function="rmse"``.
        max_evaluations: Maximum number of expressions to evaluate. Values less than 0 mean no limit.
        ground_truth: The ground truth expression, as a list of tokens in infix notation, a
            [Node][SRToolkit.utils.expression_tree.Node] tree, or a numpy array of behavior matrix
            (see [create_behavior_matrix][SRToolkit.utils.measures.create_behavior_matrix]).
            Numpy array is only applicable when ``ranking_function="bed"``.
        original_equation: Human-readable string of the original equation (e.g. ``"z = x + y"``).
        success_threshold: Error threshold below which an expression is considered successful. If ``None``,
            no threshold is applied.
        seed: Random seed for reproducibility. ``None`` means no seed is set.
        dataset_metadata: Optional dictionary of metadata about the dataset (e.g. citation, variable names).
        dataset_name: Name for this dataset. Defaults to ``"unnamed"``.
        samplers: Optional list of [Sampler][SRToolkit.dataset.sampling.Sampler]
            instances (one per input variable). The built-in
            [LogUniformSampling][SRToolkit.dataset.sampling.LogUniformSampling],
            [UniformSampling][SRToolkit.dataset.sampling.UniformSampling], and
            [IntegerUniformSampling][SRToolkit.dataset.sampling.IntegerUniformSampling]
            implement this interface.
        benchmark: Optional benchmark name (e.g. ``"feynman"``). Required for serialisation
            via [to_dict][SRToolkit.dataset.sr_dataset.SR_dataset.to_dict].
        version: Optional version string (e.g. ``"1.0.0"``). Required for serialisation.
        **kwargs: Optional estimation settings passed to
            [SR_evaluator][SRToolkit.evaluation.sr_evaluator.SR_evaluator].
            Supported keys: ``method``, ``tol``, ``gtol``, ``max_iter``, ``constant_bounds``,
            ``initialization``, ``max_constants``, ``max_expr_length``, ``num_points_sampled``,
            ``bed_X``, ``num_consts_sampled``, ``domain_bounds``.
    """
    self.X = X
    self.symbol_library = symbol_library
    self.y = y
    self.max_evaluations = max_evaluations
    self.success_threshold = success_threshold
    self.ranking_function = ranking_function
    self.ground_truth = ground_truth
    self.original_equation = original_equation
    self.kwargs = kwargs
    self.dataset_name = dataset_name

    # See if symbols contain a symbol for constants
    symbols_metadata = self.symbol_library.symbols.values()
    self.contains_constants = any([symbol["type"] == "const" for symbol in symbols_metadata])

    self.seed = seed
    self.dataset_metadata = dataset_metadata
    self.samplers = samplers

    # Cache / serialisation metadata
    self.benchmark = benchmark
    self.version = version
    # Origin of the cached data (UrlSource / SampleSource / None). Not a constructor
    # argument: it is set by the factory methods (from_dict, from_samplers) and by
    # SR_benchmark.add_dataset. End users select an origin through those entry points.
    self.data_source: Optional[DataSource] = None

evaluate_approach

evaluate_approach(sr_approach: SR_approach, num_experiments: int = 1, top_k: int = 20, initial_seed: Optional[int] = None, results: Optional[SR_results] = None, callbacks: Optional[Union[SRCallbacks, CallbackDispatcher, List[SRCallbacks]]] = None, verbose: bool = True, adaptation_path: Optional[str] = None) -> SR_results

Evaluates an SR approach on this dataset.

Examples:

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y")
>>> results = dataset.evaluate_approach(my_approach, num_experiments=5)

Parameters:

Name	Type	Description	Default
sr_approach	SR_approach	The SR approach to evaluate.	required
num_experiments	int	Number of independent experiments (runs) to perform.	1
top_k	int	Number of top expressions to retain per experiment.	20
initial_seed	Optional[int]	Seed for random number generation. If None, the dataset seed is used.	None
results	Optional[SR_results]	Existing SR_results object to append results to. If None, a new one is created.	None
callbacks	Optional[Union[SRCallbacks, CallbackDispatcher, List[SRCallbacks]]]	Optional list of SRCallbacks, SRCallbacks, or CallbackDispatcher for monitoring and controlling the search.	None
verbose	bool	If True, prints progress for each experiment.	True
adaptation_path	Optional[str]	Path to save/load the adapted state for approaches with adaptation_scope="once". If the file already exists it is loaded directly, skipping adaptation. If it does not exist, the approach is adapted and the state is saved to this path. If None, adaptation runs without saving.	None

Returns:

Type	Description
SR_results	An SR_results object containing results from all experiments.

Source code in SRToolkit/dataset/sr_dataset.py

def evaluate_approach(
    self,
    sr_approach: SR_approach,
    num_experiments: int = 1,
    top_k: int = 20,
    initial_seed: Optional[int] = None,
    results: Optional[SR_results] = None,
    callbacks: Optional[Union[SRCallbacks, CallbackDispatcher, List[SRCallbacks]]] = None,
    verbose: bool = True,
    adaptation_path: Optional[str] = None,
) -> SR_results:
    """
    Evaluates an SR approach on this dataset.

    Examples:
        >>> X = np.array([[1, 2], [3, 4], [5, 6]])
        >>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
        ...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y")
        >>> results = dataset.evaluate_approach(my_approach, num_experiments=5)  # doctest: +SKIP

    Args:
        sr_approach: The SR approach to evaluate.
        num_experiments: Number of independent experiments (runs) to perform.
        top_k: Number of top expressions to retain per experiment.
        initial_seed: Seed for random number generation. If ``None``, the dataset seed is used.
        results: Existing [SR_results][SRToolkit.evaluation.sr_evaluator.SR_results] object to append
            results to. If ``None``, a new one is created.
        callbacks: Optional list of [SRCallbacks][SRToolkit.evaluation.callbacks.SRCallbacks], [SRCallbacks][SRToolkit.evaluation.callbacks.SRCallbacks], or
            [CallbackDispatcher][SRToolkit.evaluation.callbacks.CallbackDispatcher] for monitoring
            and controlling the search.
        verbose: If ``True``, prints progress for each experiment.
        adaptation_path: Path to save/load the adapted state for approaches with
            ``adaptation_scope="once"``. If the file already exists it is loaded directly,
            skipping adaptation. If it does not exist, the approach is adapted and the state
            is saved to this path. If ``None``, adaptation runs without saving.

    Returns:
        An [SR_results][SRToolkit.evaluation.sr_evaluator.SR_results] object containing results from all experiments.
    """
    if initial_seed is None:
        seed = self.seed
    else:
        seed = initial_seed

    if results is None:
        results = SR_results()

    if isinstance(callbacks, SRCallbacks):
        dispatcher = CallbackDispatcher(callbacks=[callbacks])
        callbacks = dispatcher
    elif isinstance(callbacks, list):
        if len(callbacks) == 0:
            callbacks = None
        else:
            callbacks = CallbackDispatcher(callbacks=callbacks)

    dataset_name = self.dataset_name

    if sr_approach.adaptation_scope == "once":
        if adaptation_path is not None and os.path.exists(adaptation_path):
            sr_approach.load_adapted_state(adaptation_path)
        else:
            sr_approach.adapt(self.X, self.symbol_library)
            if adaptation_path is not None:
                dir_name = os.path.dirname(adaptation_path)
                if dir_name:
                    os.makedirs(dir_name, exist_ok=True)
                sr_approach.save_adapted_state(adaptation_path)

    for experiment in range(num_experiments):
        if verbose:
            print(f"Running experiment {experiment + 1}/{num_experiments}")
        if seed is not None:
            seed += 1

        event = ExperimentEvent(
            dataset_name=dataset_name,
            approach_name=sr_approach.name,
            success_threshold=self.success_threshold,
            max_evaluations=self.max_evaluations,
            seed=seed,
        )
        if callbacks is not None:
            callbacks.on_experiment_start(event)

        sr_approach.prepare()

        if sr_approach.adaptation_scope == "experiment":
            sr_approach.adapt(self.X, self.symbol_library)

        evaluator = self.create_evaluator(seed=seed)
        evaluator._experiment_id = f"{dataset_name}_{sr_approach.name}_{seed}"
        evaluator.register_callbacks(callbacks)
        start_time = time.monotonic()
        sr_approach.search(evaluator, seed)
        elapsed = time.monotonic() - start_time
        experiment_results = evaluator.get_results(sr_approach.name, top_k)
        experiment_results.results[-1].wall_time = elapsed
        results += experiment_results

        if callbacks is not None:
            callbacks.on_experiment_end(event, results.results[-1])
    return results

create_evaluator

create_evaluator(metadata: Optional[Dict[str, Any]] = None, seed: Optional[int] = None) -> SR_evaluator

Creates an instance of the SR_evaluator class from this dataset.

Examples:

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y", success_threshold=1e-6)
>>> evaluator = dataset.create_evaluator()
>>> float(evaluator.evaluate_expr(["sin", "(", "X_0", ")"]))
8.05645397...
>>> float(evaluator.evaluate_expr(["X_1", "+", "X_0"]))
0.0...

Parameters:

Name	Type	Description	Default
metadata	Optional[Dict[str, Any]]	Optional dictionary of metadata to attach to the evaluator (e.g. model name, seed). Dataset metadata is merged in automatically.	None
seed	Optional[int]	Seed for the random number generator. If None, the dataset seed is used.	None

Returns:

Type	Description
SR_evaluator	A configured SR_evaluator ready to evaluate expressions against this dataset.

Raises:

Type	Description
Exception	If SR_evaluator cannot be instantiated with the current dataset settings.

Source code in SRToolkit/dataset/sr_dataset.py

def create_evaluator(self, metadata: Optional[Dict[str, Any]] = None, seed: Optional[int] = None) -> SR_evaluator:
    """
    Creates an instance of the SR_evaluator class from this dataset.

    Examples:
        >>> X = np.array([[1, 2], [3, 4], [5, 6]])
        >>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
        ...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y", success_threshold=1e-6)
        >>> evaluator = dataset.create_evaluator()
        >>> float(evaluator.evaluate_expr(["sin", "(", "X_0", ")"]))  # doctest: +ELLIPSIS
        8.05645397...
        >>> float(evaluator.evaluate_expr(["X_1", "+", "X_0"]))  # doctest: +ELLIPSIS
        0.0...

    Args:
        metadata: Optional dictionary of metadata to attach to the evaluator (e.g. model name, seed).
            Dataset metadata is merged in automatically.
        seed: Seed for the random number generator. If ``None``, the dataset seed is used.

    Returns:
        A configured [SR_evaluator][SRToolkit.evaluation.sr_evaluator.SR_evaluator] ready to evaluate expressions against this dataset.

    Raises:
        Exception: If [SR_evaluator][SRToolkit.evaluation.sr_evaluator.SR_evaluator] cannot be
            instantiated with the current dataset settings.
    """
    if metadata is None:
        metadata = dict()
    if self.dataset_metadata is not None:
        metadata["dataset_metadata"] = self.dataset_metadata
    metadata["dataset_name"] = self.dataset_name

    if seed is None:
        seed = self.seed

    return SR_evaluator(
        X=self.X,
        y=self.y,
        max_evaluations=self.max_evaluations,
        success_threshold=self.success_threshold,
        ranking_function=self.ranking_function,
        ground_truth=self.ground_truth,
        symbol_library=self.symbol_library,
        seed=seed,
        metadata=metadata,
        **self.kwargs,
    )

__str__

__str__() -> str

Returns a string describing this dataset.

The string describes the target expression, symbols that should be used, and the success threshold. It also includes any constraints that should be followed when evaluating a model on this dataset. These constraints include the maximum number of expressions to evaluate, the maximum length of the expression, and the maximum number of constants allowed in the expression. If the symbol library contains a symbol for constants, the string also includes the range of constants.

For other metadata, please refer to the attribute self.dataset_metadata.

Examples:

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y", success_threshold=1e-6)
>>> str(dataset)
'Dataset for target expression z = x + y. When evaluating your model on this dataset, you should limit your generative model to only produce expressions using the following symbols: +, -, *, /, ^, u-, sqrt, sin, cos, exp, tan, arcsin, arccos, arctan, sinh, cosh, tanh, floor, ceil, ln, log, ^-1, ^2, ^3, ^4, ^5, pi, e, C, X_0, X_1.\nExpressions will be ranked based on the RMSE ranking function.\nExpressions are deemed successful if the root mean squared error is less than 1e-06. However, we advise that you check the best performing expressions manually to ensure they are correct.\nDataset uses the default limitations (extra arguments) from the SR_evaluator.The expressions in the dataset can contain constants/free parameters.\nFor other metadata, please refer to the attribute self.dataset_metadata.'

Returns:

Type	Description
str	A string describing this dataset.

Source code in SRToolkit/dataset/sr_dataset.py

def __str__(self) -> str:
    r"""
    Returns a string describing this dataset.

    The string describes the target expression, symbols that should be used,
    and the success threshold. It also includes any constraints that should
    be followed when evaluating a model on this dataset. These constraints include the maximum
    number of expressions to evaluate, the maximum length of the expression,
    and the maximum number of constants allowed in the expression. If the
    symbol library contains a symbol for constants, the string also includes
    the range of constants.

    For other metadata, please refer to the attribute self.dataset_metadata.

    Examples:
        >>> X = np.array([[1, 2], [3, 4], [5, 6]])
        >>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
        ...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y", success_threshold=1e-6)
        >>> str(dataset)
        'Dataset for target expression z = x + y. When evaluating your model on this dataset, you should limit your generative model to only produce expressions using the following symbols: +, -, *, /, ^, u-, sqrt, sin, cos, exp, tan, arcsin, arccos, arctan, sinh, cosh, tanh, floor, ceil, ln, log, ^-1, ^2, ^3, ^4, ^5, pi, e, C, X_0, X_1.\nExpressions will be ranked based on the RMSE ranking function.\nExpressions are deemed successful if the root mean squared error is less than 1e-06. However, we advise that you check the best performing expressions manually to ensure they are correct.\nDataset uses the default limitations (extra arguments) from the SR_evaluator.The expressions in the dataset can contain constants/free parameters.\nFor other metadata, please refer to the attribute self.dataset_metadata.'

    Returns:
        A string describing this dataset.
    """
    description = f"Dataset for target expression {self.original_equation}."
    description += (
        f" When evaluating your model on this dataset, you should limit your generative model to only "
        f"produce expressions using the following symbols: {str(self.symbol_library)}.\nExpressions will be "
        f"ranked based on the {self.ranking_function.upper()} ranking function.\n"
    )

    if self.success_threshold is not None:
        description += (
            "Expressions are deemed successful if the root mean squared error is less than "
            f"{self.success_threshold}. However, we advise that you check the best performing "
            f"expressions manually to ensure they are correct.\n"
        )

    if len(self.kwargs) == 0:
        description += "Dataset uses the default limitations (extra arguments) from the SR_evaluator."
    else:
        limitations = "Non default limitations (extra arguments) from the SR_evaluators are:"
        for key, value in self.kwargs.items():
            limitations += f" {key}={value}, "
        limitations = limitations[:-2] + ".\n"
        description += limitations

    if self.contains_constants:
        description += "The expressions in the dataset can contain constants/free parameters.\n"

    description += "For other metadata, please refer to the attribute self.dataset_metadata."

    return description

resample

resample(n: int, seed: Optional[int] = None) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]

Generate fresh data by applying the stored samplers to produce new inputs.

For ranking_function="rmse", the ground truth expression is also evaluated and (X, y) is returned. For ranking_function="bed", only X is returned.

Parameters:

Name	Type	Description	Default
n	int	Number of samples to generate.	required
seed	Optional[int]	Random seed for reproducibility. If None, no seed is set.	None

Returns:

Type	Description
Union[ndarray, Tuple[ndarray, ndarray]]	For RMSE: a tuple (X, y) with shapes (n, n_features) and (n,).
Union[ndarray, Tuple[ndarray, ndarray]]	For BED: a single array X with shape (n, n_features).

Raises:

Type	Description
ValueError	If samplers is None, or if ranking_function="rmse" and ground_truth is None or a behaviour array.

Source code in SRToolkit/dataset/sr_dataset.py

def resample(self, n: int, seed: Optional[int] = None) -> Union[np.ndarray, Tuple[np.ndarray, np.ndarray]]:
    """
    Generate fresh data by applying the stored samplers to produce new inputs.

    For ``ranking_function="rmse"``, the ground truth expression is also evaluated and
    ``(X, y)`` is returned. For ``ranking_function="bed"``, only ``X`` is returned.

    Args:
        n: Number of samples to generate.
        seed: Random seed for reproducibility. If ``None``, no seed is set.

    Returns:
        For RMSE: a tuple ``(X, y)`` with shapes ``(n, n_features)`` and ``(n,)``.
        For BED: a single array ``X`` with shape ``(n, n_features)``.

    Raises:
        ValueError: If ``samplers`` is ``None``, or if ``ranking_function="rmse"`` and
            ``ground_truth`` is ``None`` or a behaviour array.
    """
    if self.samplers is None:
        raise ValueError(
            f"[SR_dataset.resample] Dataset '{self.dataset_name}' has no samplers defined. "
            "Provide samplers when constructing the dataset."
        )
    if seed is not None:
        np.random.seed(seed)
    X = np.column_stack([s(n) for s in self.samplers])
    if self.ranking_function == "bed":
        return X
    if self.ground_truth is None or isinstance(self.ground_truth, np.ndarray):
        raise ValueError(
            f"[SR_dataset.resample] Dataset '{self.dataset_name}' has no token-list ground truth — "
            "cannot evaluate y. ground_truth must be a list of tokens or a Node."
        )

    # Generating y requires a fully specified ground truth: a free constant ("const"
    # token, e.g. "C") has no value to evaluate against. Reject it with a clear error
    # rather than passing an empty constant array (which the compiled Cython backend
    # would read out of bounds, silently producing garbage y values).
    if isinstance(self.ground_truth, list):
        tokens = self.ground_truth
    else:
        tokens = self.ground_truth.to_list(self.symbol_library)
    const_symbols = set(self.symbol_library.get_symbols_of_type("const"))
    if any(tok in const_symbols for tok in tokens):
        raise ValueError(
            f"[SR_dataset.resample] Dataset '{self.dataset_name}' ground truth contains a free "
            f"constant ({', '.join(sorted(const_symbols))}), which has no value to evaluate. "
            "Sampler-based generation requires a fully specified ground truth — replace the "
            "free constant with a concrete value (e.g. a literal) or a sub-expression."
        )

    f = compile_expr(self.ground_truth, self.symbol_library)
    y = f(X, np.array([]))
    return X, y

resample_inplace

resample_inplace(n: int, seed: Optional[int] = None) -> SR_dataset

Resample via the stored samplers and write the result back into self.X/self.y.

A convenience wrapper around resample that handles the RMSE (X, y) vs BED X-only return shapes. For BED datasets, self.y is set to None.

Parameters:

Name	Type	Description	Default
n	int	Number of samples to generate.	required
seed	Optional[int]	Random seed for reproducibility. If None, no seed is set.	None

Returns:

Type	Description
SR_dataset	self, with X (and y) replaced by the freshly sampled data.

Raises:

Type	Description
ValueError	Propagated from resample if samplers is None or the ground truth cannot produce y.

Source code in SRToolkit/dataset/sr_dataset.py

def resample_inplace(self, n: int, seed: Optional[int] = None) -> "SR_dataset":
    """
    Resample via the stored samplers and write the result back into ``self.X``/``self.y``.

    A convenience wrapper around [resample][SRToolkit.dataset.sr_dataset.SR_dataset.resample]
    that handles the RMSE ``(X, y)`` vs BED ``X``-only return shapes. For BED datasets,
    ``self.y`` is set to ``None``.

    Args:
        n: Number of samples to generate.
        seed: Random seed for reproducibility. If ``None``, no seed is set.

    Returns:
        ``self``, with ``X`` (and ``y``) replaced by the freshly sampled data.

    Raises:
        ValueError: Propagated from [resample][SRToolkit.dataset.sr_dataset.SR_dataset.resample]
            if ``samplers`` is ``None`` or the ground truth cannot produce ``y``.
    """
    result = self.resample(n, seed=seed)
    if isinstance(result, tuple):
        self.X, self.y = result
    else:
        self.X, self.y = result, None
    return self

to_dict

to_dict() -> dict

Creates a JSON-safe dictionary representation of this dataset.

The data arrays are not embedded in the dictionary — they live in the data cache (see data_cache). Use from_dict to reconstruct the full dataset including data.

When data_source is None the in-memory arrays are the only copy of the data, so calling this method writes them into the cache version directory (via the private _persist_to_cache helper) so the returned config stays reloadable. For SampleSource / UrlSource datasets there are no arrays to write and the call has no filesystem side effects.

Examples:

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y",
...     success_threshold=1e-6, benchmark="test", version="1.0.0")
>>> d = dataset.to_dict()
>>> d["format_version"]
2
>>> d["benchmark"]
'test'

Returns:

Type	Description
dict	A JSON-safe dictionary representing this dataset's configuration.

Raises:

Type	Description
ValueError	If benchmark or version is None (both are required for serialisation so the cache layer can locate the data).
ValueError	If ground_truth is not the correct type.

Source code in SRToolkit/dataset/sr_dataset.py

def to_dict(self) -> dict:
    r"""
    Creates a JSON-safe dictionary representation of this dataset.

    The data arrays are not embedded in the dictionary — they live in the
    data cache (see [data_cache][SRToolkit.dataset.data_cache]). Use
    [from_dict][SRToolkit.dataset.sr_dataset.SR_dataset.from_dict] to
    reconstruct the full dataset including data.

    When ``data_source`` is ``None`` the in-memory arrays are the only copy of the
    data, so calling this method writes them into the cache version directory (via
    the private ``_persist_to_cache`` helper) so the returned config stays
    reloadable. For
    [SampleSource][SRToolkit.dataset.data_source.SampleSource] /
    [UrlSource][SRToolkit.dataset.data_source.UrlSource] datasets there are no arrays
    to write and the call has no filesystem side effects.

    Examples:
        >>> X = np.array([[1, 2], [3, 4], [5, 6]])
        >>> dataset = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
        ...     y=np.array([3, 7, 11]), max_evaluations=10000, original_equation="z = x + y",
        ...     success_threshold=1e-6, benchmark="test", version="1.0.0")
        >>> d = dataset.to_dict()
        >>> d["format_version"]
        2
        >>> d["benchmark"]
        'test'

    Returns:
        A JSON-safe dictionary representing this dataset's configuration.

    Raises:
        ValueError: If ``benchmark`` or ``version`` is ``None`` (both are required for
            serialisation so the cache layer can locate the data).
        ValueError: If ``ground_truth`` is not the correct type.
    """
    if self.benchmark is None:
        raise ValueError("[SR_dataset.to_dict] 'benchmark' is None. Set self.benchmark before serialising.")
    if self.version is None:
        raise ValueError("[SR_dataset.to_dict] 'version' is None. Set self.version before serialising.")

    # When there is no DataSource, the in-memory arrays are the only copy of the
    # data. Persist them into the cache so the returned config is actually
    # reloadable by from_dict (which materialises from the cache). For sample/url
    # sources there are no arrays to write, so to_dict stays side-effect-free.
    if self.data_source is None and self.X is not None:
        self._persist_to_cache()

    # Serialise kwargs, converting ndarray values
    kwargs_out = {}
    for k, v in self.kwargs.items():
        if isinstance(v, np.ndarray):
            kwargs_out[k] = v.tolist()
        else:
            kwargs_out[k] = v

    # Serialise ground truth
    if self.ground_truth is None:
        ground_truth_out = None
    elif isinstance(self.ground_truth, list):
        ground_truth_out = self.ground_truth
    elif isinstance(self.ground_truth, Node):
        ground_truth_out = self.ground_truth.to_list()
    elif isinstance(self.ground_truth, np.ndarray):
        # ndarray ground truth lives in the cache as a separate file
        ground_truth_out = None
    else:
        raise ValueError("[SR_dataset.to_dict] Ground truth must be either a list, a Node, or a numpy array")

    return {
        "format_version": 2,
        "dataset_name": self.dataset_name,
        "benchmark": self.benchmark,
        "version": self.version,
        "symbol_library": self.symbol_library.to_dict(),
        "ranking_function": self.ranking_function,
        "max_evaluations": self.max_evaluations,
        "success_threshold": self.success_threshold,
        "original_equation": self.original_equation,
        "seed": self.seed,
        "dataset_metadata": self.dataset_metadata,
        "kwargs": kwargs_out,
        "samplers": [s.to_dict() for s in self.samplers] if self.samplers is not None else None,
        "ground_truth": ground_truth_out,
        "data_source": self.data_source.to_dict() if self.data_source is not None else None,
    }

from_dict classmethod

from_dict(d: Union[dict, str, Path]) -> SR_dataset

Creates an instance of the SR_dataset class from its dictionary representation.

If d is a string or Path, it is treated as a JSON file path and read from disk. To load a self-contained .zip archive (written by to_archive) use from_archive instead. The data arrays are loaded from the data cache (or materialised on demand).

Examples:

>>> import tempfile, os, json
>>> from SRToolkit.dataset.sampling import UniformSampling
>>> from SRToolkit.dataset.data_source import SampleSource
>>> X = np.array([[1, 2], [3, 4], [5, 6]], dtype=float)
>>> ds = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
...     y=np.array([3, 7, 11], dtype=float), max_evaluations=10000, original_equation="z = x + y",
...     success_threshold=1e-6, benchmark="test_bench", version="1.0.0",
...     samplers=[UniformSampling(0, 5), UniformSampling(0, 5)])
>>> ds.data_source = SampleSource(n_samples=3, seed=0)
>>> d = ds.to_dict()
>>> ds2 = SR_dataset.from_dict(d)
>>> ds2.benchmark
'test_bench'

Parameters:

Name	Type	Description	Default
d	Union[dict, str, Path]	Dictionary produced by to_dict, or a path to a JSON file containing such a dictionary.	required

Returns:

Type	Description
SR_dataset	A new SR_dataset instance.

Raises:

Type	Description
ValueError	If the format_version is not 2.
FileNotFoundError	If the cached data file does not exist and data_source is None.

Source code in SRToolkit/dataset/sr_dataset.py

@classmethod
def from_dict(cls, d: Union[dict, str, Path]) -> "SR_dataset":
    """
    Creates an instance of the SR_dataset class from its dictionary representation.

    If ``d`` is a string or Path, it is treated as a JSON file path and read
    from disk. To load a self-contained ``.zip`` archive (written by
    [to_archive][SRToolkit.dataset.sr_dataset.SR_dataset.to_archive]) use
    [from_archive][SRToolkit.dataset.sr_dataset.SR_dataset.from_archive] instead.
    The data arrays are loaded from the data cache (or materialised on demand).

    Examples:
        >>> import tempfile, os, json
        >>> from SRToolkit.dataset.sampling import UniformSampling
        >>> from SRToolkit.dataset.data_source import SampleSource
        >>> X = np.array([[1, 2], [3, 4], [5, 6]], dtype=float)
        >>> ds = SR_dataset(X, SymbolLibrary.default_symbols(2), ground_truth=["X_0", "+", "X_1"],
        ...     y=np.array([3, 7, 11], dtype=float), max_evaluations=10000, original_equation="z = x + y",
        ...     success_threshold=1e-6, benchmark="test_bench", version="1.0.0",
        ...     samplers=[UniformSampling(0, 5), UniformSampling(0, 5)])
        >>> ds.data_source = SampleSource(n_samples=3, seed=0)
        >>> d = ds.to_dict()
        >>> ds2 = SR_dataset.from_dict(d)
        >>> ds2.benchmark
        'test_bench'

    Args:
        d: Dictionary produced by
            [to_dict][SRToolkit.dataset.sr_dataset.SR_dataset.to_dict], or a path to a
            JSON file containing such a dictionary.

    Returns:
        A new [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] instance.

    Raises:
        ValueError: If the ``format_version`` is not 2.
        FileNotFoundError: If the cached data file does not exist and ``data_source`` is
            ``None``.
    """
    if isinstance(d, (str, Path)):
        if str(d).endswith(".zip"):
            raise ValueError(
                "[SR_dataset.from_dict] Received a '.zip' path. Load self-contained "
                "archives with SR_dataset.from_archive(path) instead."
            )
        with open(d) as f:
            dd = json.load(f)
    else:
        dd = dict(d)

    # Apply bundle relocation if needed
    dd = _auto_bind(dd)

    fmt = dd.get("format_version", 1)
    if fmt == 1:
        # Legacy format — delegate to old-style loading
        return cls._from_dict_v1(dd)
    if fmt != 2:
        raise ValueError(f"[SR_dataset.from_dict] Unsupported format_version: {fmt!r}. Expected 2.")

    benchmark = dd["benchmark"]
    version = dd["version"]
    dataset_name = dd["dataset_name"]

    cache_path = data_cache.resolve(benchmark, version, dataset_name, dd)

    data = np.load(str(cache_path))
    X = data["X"]
    y = data["y"] if "y" in data else None

    # Check for separate ground-truth array file
    gt_path = cache_path.parent / f"{dataset_name}_gt.npy"
    if gt_path.exists():
        ground_truth = np.load(str(gt_path))
    else:
        ground_truth = dd.get("ground_truth")

    kwargs = dict(dd.get("kwargs") or {})
    if "bed_X" in kwargs and kwargs["bed_X"] is not None:
        kwargs["bed_X"] = np.array(kwargs["bed_X"])

    samplers = None
    if dd.get("samplers") is not None:
        samplers = [Sampler.from_dict(s) for s in dd["samplers"]]

    dataset = cls(
        X,
        SymbolLibrary.from_dict(dd["symbol_library"]),
        ranking_function=dd["ranking_function"],
        y=y,
        max_evaluations=dd["max_evaluations"],
        ground_truth=ground_truth,
        original_equation=dd["original_equation"],
        success_threshold=dd["success_threshold"],
        seed=dd["seed"],
        dataset_metadata=dd.get("dataset_metadata"),
        dataset_name=dataset_name,
        samplers=samplers,
        benchmark=benchmark,
        version=version,
        **kwargs,
    )
    dataset.data_source = DataSource.from_dict(dd.get("data_source"))
    return dataset

to_archive

to_archive(path: Union[str, Path]) -> None

Write this dataset (config + data) to a self-contained .zip archive.

The archive mirrors the per-dataset layout of SR_benchmark.to_archive and contains:

• dataset.json: this dataset's configuration dict (see to_dict).
• data/<dataset_name>.npz: the X (and y for RMSE) arrays.
• data/<dataset_name>_gt.npy: ground-truth behaviour array, only when ground_truth is a numpy array (a bed behaviour matrix).

Load it back with from_archive, or from a URL with from_url.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Destination path for the archive. A non-.zip suffix triggers a warning but is still written as a ZIP archive.	required

Raises:

Type	Description
ValueError	If benchmark or version is None (both are required so the cache layer can locate the data on load — raised by to_dict).

Source code in SRToolkit/dataset/sr_dataset.py

def to_archive(self, path: "Union[str, Path]") -> None:
    """
    Write this dataset (config + data) to a self-contained ``.zip`` archive.

    The archive mirrors the per-dataset layout of
    [SR_benchmark.to_archive][SRToolkit.dataset.sr_benchmark.SR_benchmark.to_archive]
    and contains:

    - ``dataset.json``: this dataset's configuration dict (see
      [to_dict][SRToolkit.dataset.sr_dataset.SR_dataset.to_dict]).
    - ``data/<dataset_name>.npz``: the ``X`` (and ``y`` for RMSE) arrays.
    - ``data/<dataset_name>_gt.npy``: ground-truth behaviour array, only when
      ``ground_truth`` is a numpy array (a ``bed`` behaviour matrix).

    Load it back with
    [from_archive][SRToolkit.dataset.sr_dataset.SR_dataset.from_archive], or
    from a URL with
    [from_url][SRToolkit.dataset.sr_dataset.SR_dataset.from_url].

    Args:
        path: Destination path for the archive. A non-``.zip`` suffix triggers a
            warning but is still written as a ZIP archive.

    Raises:
        ValueError: If ``benchmark`` or ``version`` is ``None`` (both are required
            so the cache layer can locate the data on load — raised by
            [to_dict][SRToolkit.dataset.sr_dataset.SR_dataset.to_dict]).
    """
    path = Path(path)
    if path.suffix.lower() != ".zip":
        warnings.warn(
            f"[SR_dataset.to_archive] path '{path}' does not end in '.zip'. "
            "The file will be a ZIP archive regardless of the extension.",
            stacklevel=2,
        )

    # to_dict validates that benchmark/version are set.
    dataset_json = json.dumps(self.to_dict(), indent=2)
    name = self.dataset_name

    with zipfile.ZipFile(str(path), "w", compression=zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("dataset.json", dataset_json)

        buf = io.BytesIO()
        if self.y is not None:
            np.savez(buf, X=self.X, y=self.y)
        else:
            np.savez(buf, X=self.X)
        zf.writestr(f"data/{name}.npz", buf.getvalue())

        if isinstance(self.ground_truth, np.ndarray):
            gt_buf = io.BytesIO()
            np.save(gt_buf, self.ground_truth)
            zf.writestr(f"data/{name}_gt.npy", gt_buf.getvalue())

from_archive classmethod

from_archive(path: Union[str, Path]) -> SR_dataset

Load a dataset from a self-contained .zip archive.

This is the counterpart to to_archive: it reads dataset.json from the archive, extracts the bundled data/*.npz (and any _gt.npy) into the data cache, and reconstructs the dataset from them. Unlike from_dict, no data_source materialisation is needed — the data travels inside the archive.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Path to a .zip archive written by to_archive.	required

Returns:

Type	Description
SR_dataset	A new SR_dataset instance.

Source code in SRToolkit/dataset/sr_dataset.py

@classmethod
def from_archive(cls, path: "Union[str, Path]") -> "SR_dataset":
    """
    Load a dataset from a self-contained ``.zip`` archive.

    This is the counterpart to
    [to_archive][SRToolkit.dataset.sr_dataset.SR_dataset.to_archive]: it reads
    ``dataset.json`` from the archive, extracts the bundled ``data/*.npz`` (and any
    ``_gt.npy``) into the data cache, and reconstructs the dataset from them. Unlike
    [from_dict][SRToolkit.dataset.sr_dataset.SR_dataset.from_dict], no
    ``data_source`` materialisation is needed — the data travels inside the archive.

    Args:
        path: Path to a ``.zip`` archive written by
            [to_archive][SRToolkit.dataset.sr_dataset.SR_dataset.to_archive].

    Returns:
        A new [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] instance.
    """
    with zipfile.ZipFile(str(path), "r") as zf:
        d = json.loads(zf.read("dataset.json"))

    benchmark = d["benchmark"]
    version = d["version"]

    # Extract data/*.npz (and any _gt.npy) into the cache version directory.
    data_cache.import_archive(Path(path), benchmark, version)

    # The cache is now populated, so no source needs to materialise it.
    d["data_source"] = None
    return cls.from_dict(d)

from_url classmethod

from_url(url: str) -> SR_dataset

Download a self-contained .zip archive from a URL and load it.

This is the remote counterpart to from_archive: the archive is downloaded to a temporary file and then loaded exactly as from_archive would. The url must point at an archive written by to_archive (a dataset.json plus a data/ directory) — not a bare .npz/data zip (that is what UrlSource is for).

Parameters:

Name	Type	Description	Default
url	str	URL of a .zip archive written by to_archive.	required

Returns:

Type	Description
SR_dataset	A new SR_dataset instance.

Source code in SRToolkit/dataset/sr_dataset.py

@classmethod
def from_url(cls, url: str) -> "SR_dataset":
    """
    Download a self-contained ``.zip`` archive from a URL and load it.

    This is the remote counterpart to
    [from_archive][SRToolkit.dataset.sr_dataset.SR_dataset.from_archive]: the archive
    is downloaded to a temporary file and then loaded exactly as ``from_archive``
    would. The ``url`` must point at an archive written by
    [to_archive][SRToolkit.dataset.sr_dataset.SR_dataset.to_archive] (a
    ``dataset.json`` plus a ``data/`` directory) — not a bare ``.npz``/data zip
    (that is what [UrlSource][SRToolkit.dataset.data_source.UrlSource] is for).

    Args:
        url: URL of a ``.zip`` archive written by
            [to_archive][SRToolkit.dataset.sr_dataset.SR_dataset.to_archive].

    Returns:
        A new [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] instance.
    """
    with urlopen(url) as response:
        data = response.read()

    tmp = tempfile.NamedTemporaryFile(suffix=".zip", delete=False)
    try:
        tmp.write(data)
        tmp.close()
        return cls.from_archive(tmp.name)
    finally:
        os.unlink(tmp.name)

from_samplers classmethod

from_samplers(ground_truth: Union[List[str], Node], samplers: List[Sampler], symbol_library: Optional[SymbolLibrary] = None, n_samples: int = 10000, seed: Optional[int] = None, ranking_function: str = 'rmse', original_equation: Optional[str] = None, success_threshold: Optional[float] = None, max_evaluations: int = -1, dataset_name: str = 'unnamed', dataset_metadata: Optional[dict] = None, benchmark: Optional[str] = None, version: Optional[str] = None, **kwargs: Unpack[EstimationSettings]) -> SR_dataset

Build a dataset from just a ground-truth expression and per-variable samplers.

This is the convenience constructor for the common case where you have the expression you want to recover and a sampling spec for its inputs, but no data arrays yet. The inputs X are drawn from samplers (one per variable) and, for ranking_function="rmse", the targets y are produced by evaluating ground_truth on them. The result carries a SampleSource, so it round-trips via to_dict, regenerates with refresh, and resamples with resample.

Examples:

>>> from SRToolkit.dataset.sampling import UniformSampling
>>> ds = SR_dataset.from_samplers(["X_0", "+", "X_1"],
...     [UniformSampling(0, 5), UniformSampling(0, 5)], n_samples=100, seed=0)
>>> ds.X.shape
(100, 2)
>>> ds.y.shape
(100,)
>>> ds.original_equation
'y = X_0+X_1'

Parameters:

Name	Type	Description	Default
ground_truth	Union[List[str], Node]	The ground-truth expression as a list of infix tokens or a Node. For "rmse" it is evaluated to produce y; for "bed" it is stored as the target.	required
samplers	List[Sampler]	One Sampler per input variable.	required
symbol_library	Optional[SymbolLibrary]	Token vocabulary. Defaults to default_symbols with one variable per sampler.	None
n_samples	int	Number of input rows to generate. Defaults to 10000.	10000
seed	Optional[int]	Random seed for the generation (stored on the SampleSource). None means no seed is set.	None
ranking_function	str	"rmse" or "bed".	'rmse'
original_equation	Optional[str]	Human-readable equation string. If None and ground_truth is a token list, it is auto-filled as "y = <tokens>".	None
success_threshold	Optional[float]	Error threshold for success. None means no threshold.	None
max_evaluations	int	Maximum expressions to evaluate. -1 means no limit.	-1
dataset_name	str	Name for this dataset.	'unnamed'
dataset_metadata	Optional[dict]	Optional dataset-level metadata dict.	None
benchmark	Optional[str]	Optional benchmark name (needed only for serialisation).	None
version	Optional[str]	Optional version string (needed only for serialisation).	None
**kwargs	Unpack[EstimationSettings]	Estimation settings forwarded to SR_evaluator.	{}

Returns:

Type	Description
SR_dataset	A new SR_dataset with freshly
SR_dataset	generated data.

Raises:

Type	Description
ValueError	If samplers is empty, or (via resample) if ranking_function="rmse" and ground_truth cannot be evaluated.

Source code in SRToolkit/dataset/sr_dataset.py

@classmethod
def from_samplers(
    cls,
    ground_truth: Union[List[str], Node],
    samplers: List[Sampler],
    symbol_library: Optional[SymbolLibrary] = None,
    n_samples: int = 10000,
    seed: Optional[int] = None,
    ranking_function: str = "rmse",
    original_equation: Optional[str] = None,
    success_threshold: Optional[float] = None,
    max_evaluations: int = -1,
    dataset_name: str = "unnamed",
    dataset_metadata: Optional[dict] = None,
    benchmark: Optional[str] = None,
    version: Optional[str] = None,
    **kwargs: Unpack[EstimationSettings],
) -> "SR_dataset":
    r"""
    Build a dataset from just a ground-truth expression and per-variable samplers.

    This is the convenience constructor for the common case where you have the
    expression you want to recover and a sampling spec for its inputs, but no data
    arrays yet. The inputs ``X`` are drawn from ``samplers`` (one per variable) and,
    for ``ranking_function="rmse"``, the targets ``y`` are produced by evaluating
    ``ground_truth`` on them. The result carries a
    [SampleSource][SRToolkit.dataset.data_source.SampleSource], so it round-trips via
    [to_dict][SRToolkit.dataset.sr_dataset.SR_dataset.to_dict], regenerates with
    [refresh][SRToolkit.dataset.sr_dataset.SR_dataset.refresh], and resamples with
    [resample][SRToolkit.dataset.sr_dataset.SR_dataset.resample].

    Examples:
        >>> from SRToolkit.dataset.sampling import UniformSampling
        >>> ds = SR_dataset.from_samplers(["X_0", "+", "X_1"],
        ...     [UniformSampling(0, 5), UniformSampling(0, 5)], n_samples=100, seed=0)
        >>> ds.X.shape
        (100, 2)
        >>> ds.y.shape
        (100,)
        >>> ds.original_equation
        'y = X_0+X_1'

    Args:
        ground_truth: The ground-truth expression as a list of infix tokens or a
            [Node][SRToolkit.utils.expression_tree.Node]. For ``"rmse"`` it is
            evaluated to produce ``y``; for ``"bed"`` it is stored as the target.
        samplers: One [Sampler][SRToolkit.dataset.sampling.Sampler] per input variable.
        symbol_library: Token vocabulary. Defaults to
            [default_symbols][SRToolkit.utils.symbol_library.SymbolLibrary.default_symbols]
            with one variable per sampler.
        n_samples: Number of input rows to generate. Defaults to ``10000``.
        seed: Random seed for the generation (stored on the
            [SampleSource][SRToolkit.dataset.data_source.SampleSource]). ``None`` means
            no seed is set.
        ranking_function: ``"rmse"`` or ``"bed"``.
        original_equation: Human-readable equation string. If ``None`` and
            ``ground_truth`` is a token list, it is auto-filled as ``"y = <tokens>"``.
        success_threshold: Error threshold for success. ``None`` means no threshold.
        max_evaluations: Maximum expressions to evaluate. ``-1`` means no limit.
        dataset_name: Name for this dataset.
        dataset_metadata: Optional dataset-level metadata dict.
        benchmark: Optional benchmark name (needed only for serialisation).
        version: Optional version string (needed only for serialisation).
        **kwargs: Estimation settings forwarded to
            [SR_evaluator][SRToolkit.evaluation.sr_evaluator.SR_evaluator].

    Returns:
        A new [SR_dataset][SRToolkit.dataset.sr_dataset.SR_dataset] with freshly
        generated data.

    Raises:
        ValueError: If ``samplers`` is empty, or (via
            [resample][SRToolkit.dataset.sr_dataset.SR_dataset.resample]) if
            ``ranking_function="rmse"`` and ``ground_truth`` cannot be evaluated.
    """
    if not samplers:
        raise ValueError(
            "[SR_dataset.from_samplers] 'samplers' must be a non-empty list (one sampler per input variable)."
        )
    if symbol_library is None:
        symbol_library = SymbolLibrary.default_symbols(len(samplers))
    if original_equation is None and isinstance(ground_truth, list):
        original_equation = "y = " + "".join(ground_truth)

    dataset = cls(
        np.empty((0, len(samplers))),
        symbol_library,
        ranking_function=ranking_function,
        ground_truth=ground_truth,
        original_equation=original_equation,
        success_threshold=success_threshold,
        max_evaluations=max_evaluations,
        dataset_name=dataset_name,
        dataset_metadata=dataset_metadata,
        samplers=samplers,
        benchmark=benchmark,
        version=version,
        **kwargs,
    )
    dataset.data_source = SampleSource(n_samples=n_samples, seed=seed)

    # Reuse the canonical generation path (handles rmse/bed and its validation).
    return dataset.resample_inplace(n_samples, seed=seed)

refresh

refresh() -> None

Force-refresh the cached data for this dataset by re-materialising it from data_source.

After refreshing, self.X and self.y are reloaded from the new cache.

Raises:

Type	Description
ValueError	If data_source, benchmark, or version is None.

Source code in SRToolkit/dataset/sr_dataset.py

def refresh(self) -> None:
    """
    Force-refresh the cached data for this dataset by re-materialising it
    from ``data_source``.

    After refreshing, ``self.X`` and ``self.y`` are reloaded from the new cache.

    Raises:
        ValueError: If ``data_source``, ``benchmark``, or ``version`` is ``None``.
    """
    if self.data_source is None:
        raise ValueError(
            "[SR_dataset.refresh] Cannot refresh: data_source is null. "
            "Set self.data_source to a valid source config before calling refresh()."
        )
    if self.benchmark is None or self.version is None:
        raise ValueError("[SR_dataset.refresh] Cannot refresh: 'benchmark' or 'version' is None.")

    data_cache.resolve(
        self.benchmark,
        self.version,
        self.dataset_name,
        self.to_dict(),
        force=True,
    )

    cache_path = data_cache.dataset_path(self.benchmark, self.version, self.dataset_name)
    data = np.load(str(cache_path))
    self.X = data["X"]
    if "y" in data:
        self.y = data["y"]

SRSD_Feynman

SRSD_Feynman(n_samples: int = 10000, seed: Optional[int] = 42, force_generate: bool = False)

Bases: SR_benchmark

The SRSD Feynman symbolic regression benchmark.

Contains 120 physics equations from the Feynman Symbolic Regression Dataset with per-variable sampling strategies (log-uniform, linear, or integer with sign constraints). Data is generated on first instantiation and cached as .npz files for subsequent use.

References

Matsubara et al. (2024)

Examples:

>>> benchmark = SRSD_Feynman()
>>> len(benchmark.list_datasets(verbose=False))
120

Parameters:

Name	Type	Description	Default
n_samples	int	Number of samples to generate per dataset when force_generate=True.	10000
seed	Optional[int]	Random seed used for data generation.	42
force_generate	bool	If True, generate fresh data from the stored samplers instead of downloading the pre-generated data. Defaults to False.	False

Source code in SRToolkit/dataset/srsd_feynman.py

def __init__(
    self,
    n_samples: int = 10000,
    seed: Optional[int] = 42,
    force_generate: bool = False,
):
    super().__init__("SRSD_Feynman", version="1.0.0")
    self._n_samples = n_samples
    self._seed = seed
    self._force_generate = force_generate
    self._populate()