Continuous Sampling Module¶

compute_lon ¶

compute_lon(
    func: Callable[[ndarray], float],
    dim: int,
    lower_bound: float | Sequence[float],
    upper_bound: float | Sequence[float],
    initial_points: ndarray | None = None,
    config: BasinHoppingSamplerConfig | None = None,
    lon_config: LONConfig | None = None,
    verbose: bool = False,
) -> LON

Compute a LON from an objective function.

This is the simplest way to construct a Local Optima Network. For more control, use BasinHoppingSampler directly.

Parameters:

Name	Type	Description	Default
`func`	`Callable[[ndarray], float]`	Objective function f(x) -> float to minimize, where x is in R^n_var.	required
`dim`	`int`	Number of dimensions (n_var).	required
`lower_bound`	`float \| Sequence[float]`	Lower bound (scalar or per-dimension list/array).	required
`upper_bound`	`float \| Sequence[float]`	Upper bound (scalar or per-dimension list/array).	required
`initial_points`	`ndarray \| None`	Optional array of shape (`config.n_runs`, `dim`) with starting points for each run. If `None`, points are sampled uniformly at random from the domain. Default: `None`.	`None`
`config`	`BasinHoppingSamplerConfig \| None`	Basin-Hopping sampler configuration. Uses default BasinHoppingSamplerConfig if not provided.	`None`
`lon_config`	`LONConfig \| None`	LON construction configuration. Uses default LONConfig if not provided.	`None`
`verbose`	`bool`	If True, show a progress bar during sampling. Default: `False`.	`False`

Returns:

Type	Description
`LON`	LON instance.

Example

import numpy as np def sphere(x): ... return np.sum(x**2) lon = compute_lon(sphere, dim=5, lower_bound=-5.0, upper_bound=5.0) print(f"Found {lon.n_vertices} local optima")

Source code in src/lonkit/continuous/sampling.py

def compute_lon(
    func: Callable[[np.ndarray], float],
    dim: int,
    lower_bound: float | Sequence[float],
    upper_bound: float | Sequence[float],
    initial_points: np.ndarray | None = None,
    config: BasinHoppingSamplerConfig | None = None,
    lon_config: LONConfig | None = None,
    verbose: bool = False,
) -> LON:
    """
    Compute a LON from an objective function.

    This is the simplest way to construct a Local Optima Network.
    For more control, use BasinHoppingSampler directly.

    Args:
        func: Objective function f(x) -> float to minimize, where x is in R^n_var.
        dim: Number of dimensions (n_var).
        lower_bound: Lower bound (scalar or per-dimension list/array).
        upper_bound: Upper bound (scalar or per-dimension list/array).
        initial_points: Optional array of shape (`config.n_runs`, `dim`) with starting
            points for each run. If `None`, points are sampled uniformly at
            random from the domain. Default: `None`.
        config: Basin-Hopping sampler configuration. Uses default
            BasinHoppingSamplerConfig if not provided.
        lon_config: LON construction configuration. Uses default
            LONConfig if not provided.
        verbose: If True, show a progress bar during sampling. Default: `False`.

    Returns:
        LON instance.

    Example:
        >>> import numpy as np
        >>> def sphere(x):
        ...     return np.sum(x**2)
        >>> lon = compute_lon(sphere, dim=5, lower_bound=-5.0, upper_bound=5.0)
        >>> print(f"Found {lon.n_vertices} local optima")
    """
    # Convert scalars to lists, leave sequences as-is
    lower_bounds: Sequence[float] = (
        [lower_bound] * dim if isinstance(lower_bound, int | float) else lower_bound
    )
    upper_bounds: Sequence[float] = (
        [upper_bound] * dim if isinstance(upper_bound, int | float) else upper_bound
    )

    domain = list(zip(lower_bounds, upper_bounds, strict=True))

    sampler = BasinHoppingSampler(config)
    result = sampler.sample(func, domain, initial_points=initial_points, verbose=verbose)
    return sampler.sample_to_lon(result, lon_config=lon_config)

BasinHoppingSamplerConfig `dataclass` ¶

Configuration for Basin-Hopping sampling.

Attributes:

Name	Type	Description
`n_runs`	`int`	Number of independent Basin-Hopping runs. Default: `100`.
`n_iter_no_change`	`int \| None`	Maximum number of consecutive non-improving perturbations before stopping each run. Use `None` for no limit. Setting both `n_iter_no_change` and `max_iter` to `None` will result in an error. Default: `250`.
`max_iter`	`int \| None`	Optional maximum number of total iterations (perturbation steps) per run. Use `None` for no limit. Setting both `n_iter_no_change` and `max_iter` to `None` will result in an error. Default: `None`.
`step_mode`	`StepMode`	Perturbation mode - `"percentage"` (of domain range) or `"fixed"` (absolute step size). Default: `"percentage"`.
`step_size`	`float`	Perturbation magnitude (interpretation depends on step_mode). Default: `0.1`.
`fitness_precision`	`int \| None`	Decimal precision for fitness values. Use `None` for full double precision. Passing negative values behaves the same as passing `None`. Default: `None`.
`coordinate_precision`	`int \| None`	Decimal precision for coordinate rounding and hashing. Solutions rounded to this precision are considered identical. Use `None` for full double precision (no rounding). Passing negative values behaves the same as passing `None`. Default: `5`.
`bounded`	`bool`	Whether to enforce domain bounds during perturbation. Default: `True`.
`minimizer_method`	`str \| Callable \| None`	Minimization method passed to `scipy.optimize.minimize`. Can be a string or a callable implementing a custom solver. See `scipy.optimize.minimize <https://docs.scipy.org/doc/scipy/reference/generated/scipy.optimize.minimize.html>`_ for the full list of supported methods and their options. Default: `"L-BFGS-B"`.
`minimizer_options`	`dict \| None`	Solver-specific options passed as the `options` argument to `scipy.optimize.minimize`. The available keys depend on the chosen `minimizer_method`. Use `None` to rely on scipy's defaults. Default: `None`.
`seed`	`int \| None`	Random seed for reproducibility. Default: `None`.
`n_jobs`	`int \| None`	The maximum number of concurrently running `joblib` jobs:
`Reproducibility`	is guaranteed across different ``n_jobs`` values when ``seed`` is set. Default	`1`.

Source code in src/lonkit/continuous/sampling.py

@dataclass
class BasinHoppingSamplerConfig:
    """
    Configuration for Basin-Hopping sampling.

    Attributes:
        n_runs: Number of independent Basin-Hopping runs.  Default: `100`.
        n_iter_no_change: Maximum number of consecutive non-improving perturbations before stopping each run.
            Use `None` for no limit. Setting both `n_iter_no_change` and `max_iter` to `None` will result in an error. Default: `250`.
        max_iter: Optional maximum number of total iterations (perturbation steps) per run.
            Use `None` for no limit. Setting both `n_iter_no_change` and `max_iter` to `None` will result in an error. Default: `None`.
        step_mode: Perturbation mode - `"percentage"` (of domain range)
            or `"fixed"` (absolute step size). Default: `"percentage"`.
        step_size: Perturbation magnitude (interpretation depends on step_mode). Default: `0.1`.
        fitness_precision: Decimal precision for fitness values.
            Use `None` for full double precision. Passing negative values behaves the same as passing `None`. Default: `None`.
        coordinate_precision: Decimal precision for coordinate rounding and hashing.
            Solutions rounded to this precision are considered identical.
            Use `None` for full double precision (no rounding). Passing negative values behaves the same as passing `None`. Default: `5`.
        bounded: Whether to enforce domain bounds during perturbation. Default: `True`.
        minimizer_method: Minimization method passed to ``scipy.optimize.minimize``. Can be a
            string or a callable implementing a custom solver.
            See `scipy.optimize.minimize <https://docs.scipy.org/doc/scipy/reference/generated/scipy.optimize.minimize.html>`_
            for the full list of supported methods and their options. Default: `"L-BFGS-B"`.
        minimizer_options: Solver-specific options passed as the ``options`` argument to
            ``scipy.optimize.minimize``. The available keys depend on the chosen
            ``minimizer_method``. Use ``None`` to rely on scipy's defaults.
            Default: `None`.
        seed: Random seed for reproducibility. Default: `None`.
        n_jobs: The maximum number of concurrently running ``joblib`` jobs:
        - ``1`` runs sequentially,
        - ``-1`` uses all available CPUs;
        - ``N > 1`` uses maximum of N processes;
        - ``-N`` uses maximum of ``max(1, cpu_count - N + 1)`` processes;
        - ``None`` is treated as ``1``.
        Reproducibility is guaranteed across different ``n_jobs`` values when ``seed`` is set. Default: ``1``.
    """

    n_runs: int = 100
    n_iter_no_change: int | None = 250
    max_iter: int | None = None
    step_mode: StepMode = "percentage"
    step_size: float = 0.1
    fitness_precision: int | None = None
    coordinate_precision: int | None = 5
    bounded: bool = True
    minimizer_method: str | Callable | None = "L-BFGS-B"
    minimizer_options: dict | None = None
    seed: int | None = None
    n_jobs: int | None = 1

    def __post_init__(self) -> None:
        if self.n_iter_no_change is not None and self.n_iter_no_change <= 0:
            raise ValueError("n_iter_no_change must be positive or None.")
        if self.max_iter is not None and self.max_iter <= 0:
            raise ValueError("max_iter must be positive or None.")
        if self.n_iter_no_change is None and self.max_iter is None:
            raise ValueError(
                "At least one stopping criterion must be set: n_iter_no_change and/or max_iter."
            )

BasinHoppingSampler ¶

Basin-Hopping sampler for constructing Local Optima Networks.

Basin-Hopping is a global optimization algorithm that combines random perturbations with local minimization. This implementation records transitions between local optima for LON construction.

Example

config = BasinHoppingSamplerConfig(n_runs=10, n_iter_no_change=250) sampler = BasinHoppingSampler(config) result = sampler.sample(objective_func, domain) lon = sampler.sample_to_lon(result)

Source code in src/lonkit/continuous/sampling.py

class BasinHoppingSampler:
    """
    Basin-Hopping sampler for constructing Local Optima Networks.

    Basin-Hopping is a global optimization algorithm that combines random
    perturbations with local minimization. This implementation records
    transitions between local optima for LON construction.

    Example:
        >>> config = BasinHoppingSamplerConfig(n_runs=10, n_iter_no_change=250)
        >>> sampler = BasinHoppingSampler(config)
        >>> result = sampler.sample(objective_func, domain)
        >>> lon = sampler.sample_to_lon(result)
    """

    def __init__(self, config: BasinHoppingSamplerConfig | None = None):
        self.config = config or BasinHoppingSamplerConfig()

    def _perturbation(
        self,
        x: np.ndarray,
        p: np.ndarray,
        rng: np.random.Generator,
        bounds: np.ndarray | None = None,
    ) -> np.ndarray:
        """
        Apply a random uniform perturbation to a solution.

        Args:
            x: Current solution coordinates.
            p: Per-dimension perturbation magnitude.
            rng: Random number generator for the perturbation.
            bounds: Optional (n_var, 2) array of domain bounds for clipping.

        Returns:
            Perturbed solution, clipped to bounds if ``bounded`` is enabled.
        """
        y = x + rng.uniform(low=-p, high=p)
        if self.config.bounded and bounds is not None:
            return np.clip(y, bounds[:, 0], bounds[:, 1])
        return y

    def _round_value(self, value: np.ndarray, precision: int | None) -> np.ndarray:
        """
        Round a value to the given decimal precision.

        Args:
            value: Value or array to round.
            precision: Number of decimal places. ``None`` or negative values
                skip rounding and return the input unchanged.

        Returns:
            Rounded value (or the original if precision is ``None``/negative).
        """
        if precision is None or precision < 0:
            return value
        return np.round(value, precision)

    def _hash_solution(self, x: np.ndarray) -> str:
        """
        Create hash string for a solution.

        Creates a unique identifier for a local optimum based on
        rounded coordinates.

        Args:
            x: Solution coordinates.

        Returns:
            Hash string identifying the local optimum.
        """
        x = (
            x + 0.0
        )  # Convert -0.0 to 0.0 for consistent hashing (avoids in-place mutation of input)
        precision = self.config.coordinate_precision
        formatter = str if precision is None or precision < 0 else lambda v: f"{v:.{precision}f}"
        hash_str = "_".join(formatter(v) for v in x)

        return hash_str

    def _single_bh_run(
        self,
        run: int,
        func: Callable[[np.ndarray], float],
        initial_point: np.ndarray,
        p: np.ndarray,
        bounds_array: np.ndarray | None,
        rng: np.random.Generator,
    ) -> tuple[list[dict], int]:
        """
        Execute a single Basin-Hopping run.

        Args:
            run: 1-based run index (recorded in every raw record).
            func: Objective function to minimize.
            initial_point: Starting point for this run.
            p: Per-dimension perturbation magnitude.
            bounds_array: (n_var, 2) bounds array, or ``None`` if unbounded.
            rng: Random number generator for perturbations in this run.

        Returns:
            Tuple of (records, nfev) for this run.
        """
        nfev = 0

        try:
            res = minimize(
                func,
                initial_point,
                method=self.config.minimizer_method,
                options=self.config.minimizer_options,
                bounds=bounds_array if self.config.bounded else None,
            )
        except ValueError as e:
            warnings.warn(
                f"Run {run}: initial minimize failed with ValueError: {e}. "
                f"Starting point: {initial_point}. Skipping run.",
                stacklevel=3,
            )
            return [], 0

        current_x = res.x
        current_f = res.fun
        nfev += res.nfev

        records: list[dict] = []
        iters_without_improvement = 0
        iter_index = 0

        while True:
            if self.config.max_iter is not None and iter_index >= self.config.max_iter:
                break
            if (
                self.config.n_iter_no_change is not None
                and iters_without_improvement >= self.config.n_iter_no_change
            ):
                break

            x_perturbed = self._perturbation(current_x, p, rng, bounds_array)
            try:
                res = minimize(
                    func,
                    x_perturbed,
                    method=self.config.minimizer_method,
                    options=self.config.minimizer_options,
                    bounds=bounds_array if self.config.bounded else None,
                )
            except ValueError as e:
                # L-BFGS-B can produce internal iterates that slightly
                # violate bounds, causing approx_derivative to fail.
                # Skip this perturbation and try the next one.
                warnings.warn(
                    f"Run {run}, iteration {iter_index}: minimize after perturbation "
                    f"failed with ValueError: {e}. "
                    f"Perturbed point: {x_perturbed}. Skipping perturbation.",
                    stacklevel=3,
                )
                iters_without_improvement += 1
                iter_index += 1
                continue

            new_x = res.x
            new_f = res.fun
            nfev += res.nfev

            records.append(
                {
                    "run": run,
                    "iteration": iter_index,
                    "current_x": current_x.copy(),
                    "current_f": current_f,
                    "new_x": new_x.copy(),
                    "new_f": new_f,
                    "accepted": new_f <= current_f,
                }
            )

            if self.config.n_iter_no_change is not None:
                if new_f < current_f:
                    iters_without_improvement = 0
                else:
                    iters_without_improvement += 1

            # Acceptance criterion (minimization: accept if better or equal)
            if new_f <= current_f:
                current_x = new_x.copy()
                current_f = new_f

            iter_index += 1

        return records, nfev

    def _sequential_bh(
        self,
        func: Callable[[np.ndarray], float],
        initial_points: np.ndarray,
        p: np.ndarray,
        bounds_array: np.ndarray | None,
        run_seeds: list[np.random.SeedSequence],
        progress_callback: Callable[[int, int], None] | None = None,
        verbose: bool = False,
    ) -> tuple[list[dict], int]:
        """
        Executes set of ``_single_bh_run`` calls in sequence, without using ``joblib``.

        Used when ``joblib.effective_n_jobs`` equals ``1``.
        """
        raw_records: list[dict] = []
        nfev_total = 0

        runs = range(1, self.config.n_runs + 1)
        run_iter = tqdm(runs, total=self.config.n_runs) if verbose else runs
        for run in run_iter:
            rng = np.random.default_rng(run_seeds[run - 1])
            records, nfev = self._single_bh_run(
                run, func, initial_points[run - 1], p, bounds_array, rng
            )
            if progress_callback:
                progress_callback(run, self.config.n_runs)
            raw_records.extend(records)
            nfev_total += nfev

        return raw_records, nfev_total

    def _parallel_bh(
        self,
        func: Callable[[np.ndarray], float],
        initial_points: np.ndarray,
        p: np.ndarray,
        bounds_array: np.ndarray | None,
        run_seeds: list[np.random.SeedSequence],
        effective_n_jobs: int,
        progress_callback: Callable[[int, int], None] | None = None,
        verbose: bool = False,
    ) -> tuple[list[dict], int]:
        """
        Executes set of ``_single_bh_run`` calls in parallel using ``joblib``.

        At the moment, the joblib runner uses "generator_unordered" setting,
        which makes the order of the completion and callback function calls non-deterministic.
        This does not affect the ``LON`` creation, as the order of the iterations within the run
        remains the same in the resulting array, allowing for stable sorting by run later.

        The ``joblib.Parallel`` runner needs to use the ``_run_single_bh_in_worker`` function, because
        of pickling - the pickled objects cannot share the same memory to achieve true parallelism between the
        processes, so we need to create separate ``BasinHoppingSampler`` object for each run.
        """
        parallel_runner = joblib.Parallel(
            n_jobs=effective_n_jobs,
            prefer="processes",
            return_as="generator_unordered",
        )

        parallel_results = parallel_runner(
            joblib.delayed(_run_single_bh_in_worker)(
                run + 1,
                func,
                initial_points[run],
                p,
                bounds_array,
                self.config,
                run_seeds[run],
            )
            for run in range(self.config.n_runs)
        )

        raw_records: list[dict] = []
        nfev_total = 0
        result_iter = (
            tqdm(parallel_results, total=self.config.n_runs) if verbose else parallel_results
        )
        for _, (run, records, nfev) in enumerate(result_iter, 1):
            if progress_callback:
                progress_callback(run, self.config.n_runs)
            raw_records.extend(records)
            nfev_total += nfev

        return raw_records, nfev_total

    def _basin_hopping_sampling(
        self,
        func: Callable[[np.ndarray], float],
        domain: list[tuple[float, float]],
        initial_points: np.ndarray,
        progress_callback: Callable[[int, int], None] | None = None,
        verbose: bool = False,
    ) -> tuple[list[dict], int]:
        """
        Run Basin-Hopping sampling to generate LON data.

        Runs are dispatched sequentially (``n_jobs=1``) or in parallel across
        separate processes (``n_jobs != 1``).  Results are identical for the same
        ``seed`` regardless of the ``n_jobs`` value because every run receives a
        deterministic, independent RNG seed derived from
        ``numpy.random.SeedSequence(config.seed)``.

        Args:
            func: Objective function to minimize (f: R^n_var -> R).
            domain: List of (lower, upper) bounds per dimension.
            initial_points: Array of shape (config.n_runs, n_var) with initial points.
            progress_callback: Optional callback(run, total_runs) for progress.
                Called after each run completes. Always called from the main process,
                regardless of ``n_jobs``.

        Returns:
            Tuple of (raw_records, nfev_total) where raw_records is a list of
            raw sampling records (one per perturbation step, each a dict with keys:
            run, iteration, current_x, current_f, new_x, new_f, accepted) and
            nfev_total is the total number of function evaluations.
        """
        n_var = len(domain)
        domain_array = np.array(domain)

        # Compute step size based on mode
        if self.config.step_mode == "percentage":
            p = self.config.step_size * np.abs(domain_array[:, 1] - domain_array[:, 0])
        else:
            p = self.config.step_size * np.ones(n_var)

        bounds_array = domain_array if self.config.bounded else None

        # Derive one independent, deterministic seed per run from the root seed.
        # Using SeedSequence guarantees statistical independence between run RNGs
        # and reproducibility across different n_jobs values.
        run_seeds = np.random.SeedSequence(self.config.seed).spawn(self.config.n_runs)

        effective_n_jobs = joblib.effective_n_jobs(self.config.n_jobs)

        if effective_n_jobs == 1:
            return self._sequential_bh(
                func,
                initial_points,
                p,
                bounds_array,
                run_seeds,
                progress_callback,
                verbose,
            )

        return self._parallel_bh(
            func,
            initial_points,
            p,
            bounds_array,
            run_seeds,
            effective_n_jobs,
            progress_callback,
            verbose,
        )

    def _construct_trace_data(self, raw_records: list[dict]) -> pd.DataFrame:
        """
        Construct trace data from accepted transitions in raw records. The returned
        trace DataFrame is sorted by run.

        Args:
            raw_records: List of raw sampling records from basin hopping.

        Returns:
            DataFrame with columns `[run, fit1, node1, fit2, node2]` representing
            actual transitions from current_x to new_x for each accepted move.
        """
        trace_records = []

        for rec in raw_records:
            if not rec["accepted"]:
                continue

            from_x = rec["current_x"]
            from_f = rec["current_f"]
            to_x = rec["new_x"]
            to_f = rec["new_f"]

            from_x_rounded = self._round_value(from_x, self.config.coordinate_precision)
            to_x_rounded = self._round_value(to_x, self.config.coordinate_precision)

            node1 = self._hash_solution(from_x_rounded)
            node2 = self._hash_solution(to_x_rounded)

            fit1 = self._round_value(from_f, self.config.fitness_precision)
            fit2 = self._round_value(to_f, self.config.fitness_precision)

            trace_records.append(
                {
                    "run": rec["run"],
                    "fit1": fit1,
                    "node1": node1,
                    "fit2": fit2,
                    "node2": node2,
                }
            )

        trace_df = pd.DataFrame(trace_records, columns=["run", "fit1", "node1", "fit2", "node2"])
        trace_df = trace_df.sort_values(by=["run"], kind="mergesort").reset_index(drop=True)
        return trace_df

    def _resolve_initial_points(
        self,
        initial_points: np.ndarray | None,
        domain: list[tuple[float, float]],
    ) -> np.ndarray:
        n_runs = self.config.n_runs
        n_var = len(domain)
        domain_array = np.array(domain)

        if initial_points is None:
            rng = np.random.default_rng(self.config.seed)
            return rng.uniform(domain_array[:, 0], domain_array[:, 1], size=(n_runs, n_var))

        initial_points = np.asarray(initial_points, dtype=float)

        if initial_points.ndim != 2 or initial_points.shape[1] != n_var:
            raise ValueError(
                f"initial_points must have shape (n_runs, {n_var}), got {initial_points.shape}."
            )

        if initial_points.shape[0] != n_runs:
            raise ValueError(
                f"initial_points has {initial_points.shape[0]} points, "
                f"but n_runs is {n_runs}. "
                f"These must match."
            )

        if self.config.bounded:
            lower = domain_array[:, 0]
            upper = domain_array[:, 1]
            if np.any(initial_points < lower) or np.any(initial_points > upper):
                raise ValueError(
                    "initial_points contains values outside the domain bounds. "
                    "All points must satisfy lower_bound <= x <= upper_bound "
                    "when bounded=True."
                )

        return initial_points

    def sample(
        self,
        func: Callable[[np.ndarray], float],
        domain: list[tuple[float, float]],
        initial_points: np.ndarray | None = None,
        progress_callback: Callable[[int, int], None] | None = None,
        verbose: bool = False,
    ) -> BasinHoppingResult:
        """
        Run Basin-Hopping sampling and construct trace data.

        Args:
            func: Objective function to minimize (f: R^n_var -> R).
            domain: List of (lower, upper) bounds per dimension.
            initial_points: Optional array of shape (`config.n_runs`, `n_var`) with
                starting points for each run. If `None`, points are sampled
                uniformly at random from the domain. Default: `None`.
            progress_callback: Optional callback(run, total_runs) for progress.
                Called after each run completes. Default: `None`.
            verbose:  If True, show a progress bar during sampling. Default: `False`.
        Returns:
            BasinHoppingResult: Result of the sampling run.
        """
        resolved_points = self._resolve_initial_points(initial_points, domain)

        # Collect all raw sampling data
        raw_records, nfev_total = self._basin_hopping_sampling(
            func, domain, resolved_points, progress_callback, verbose
        )

        # Construct trace data from accepted transitions
        trace_df = self._construct_trace_data(raw_records)

        return BasinHoppingResult(trace_df=trace_df, raw_records=raw_records, nfev=nfev_total)

    def sample_to_lon(
        self,
        sampler_result: BasinHoppingResult,
        lon_config: LONConfig | None = None,
    ) -> LON:
        """
        Construct a LON from a `BasinHoppingResult`.

        Convenience wrapper that passes the trace data from a sampling result
        to `LON.from_trace_data()`. Equivalent to calling
        `LON.from_trace_data(sampler_result.trace_df, config=lon_config)`.

        Args:
            sampler_result: Result returned by `sample()`.
            lon_config: LON construction configuration. If `None`, uses default
                `LONConfig`. Default: `None`.

        Returns:
            `LON` instance constructed from the sampling trace.
        """

        trace_df = sampler_result.trace_df

        if trace_df.empty:
            return LON()

        _lon_config = replace(lon_config) if lon_config is not None else LONConfig()

        if _lon_config.eq_atol is None:
            p = self.config.fitness_precision
            if p is not None and p >= 0:
                _lon_config.eq_atol = 10 ** -(p + 1)

        return LON.from_trace_data(trace_df, config=_lon_config)

sample ¶

sample(
    func: Callable[[ndarray], float],
    domain: list[tuple[float, float]],
    initial_points: ndarray | None = None,
    progress_callback: Callable[[int, int], None]
    | None = None,
    verbose: bool = False,
) -> BasinHoppingResult

Run Basin-Hopping sampling and construct trace data.

Parameters:

Name	Type	Description	Default
`func`	`Callable[[ndarray], float]`	Objective function to minimize (f: R^n_var -> R).	required
`domain`	`list[tuple[float, float]]`	List of (lower, upper) bounds per dimension.	required
`initial_points`	`ndarray \| None`	Optional array of shape (`config.n_runs`, `n_var`) with starting points for each run. If `None`, points are sampled uniformly at random from the domain. Default: `None`.	`None`
`progress_callback`	`Callable[[int, int], None] \| None`	Optional callback(run, total_runs) for progress. Called after each run completes. Default: `None`.	`None`
`verbose`	`bool`	If True, show a progress bar during sampling. Default: `False`.	`False`

Returns: BasinHoppingResult: Result of the sampling run.

Source code in src/lonkit/continuous/sampling.py

def sample(
    self,
    func: Callable[[np.ndarray], float],
    domain: list[tuple[float, float]],
    initial_points: np.ndarray | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    verbose: bool = False,
) -> BasinHoppingResult:
    """
    Run Basin-Hopping sampling and construct trace data.

    Args:
        func: Objective function to minimize (f: R^n_var -> R).
        domain: List of (lower, upper) bounds per dimension.
        initial_points: Optional array of shape (`config.n_runs`, `n_var`) with
            starting points for each run. If `None`, points are sampled
            uniformly at random from the domain. Default: `None`.
        progress_callback: Optional callback(run, total_runs) for progress.
            Called after each run completes. Default: `None`.
        verbose:  If True, show a progress bar during sampling. Default: `False`.
    Returns:
        BasinHoppingResult: Result of the sampling run.
    """
    resolved_points = self._resolve_initial_points(initial_points, domain)

    # Collect all raw sampling data
    raw_records, nfev_total = self._basin_hopping_sampling(
        func, domain, resolved_points, progress_callback, verbose
    )

    # Construct trace data from accepted transitions
    trace_df = self._construct_trace_data(raw_records)

    return BasinHoppingResult(trace_df=trace_df, raw_records=raw_records, nfev=nfev_total)

sample_to_lon ¶

sample_to_lon(
    sampler_result: BasinHoppingResult,
    lon_config: LONConfig | None = None,
) -> LON

Construct a LON from a BasinHoppingResult.

Convenience wrapper that passes the trace data from a sampling result to LON.from_trace_data(). Equivalent to calling LON.from_trace_data(sampler_result.trace_df, config=lon_config).

Parameters:

Name	Type	Description	Default
`sampler_result`	`BasinHoppingResult`	Result returned by `sample()`.	required
`lon_config`	`LONConfig \| None`	LON construction configuration. If `None`, uses default `LONConfig`. Default: `None`.	`None`

Returns:

Type	Description
`LON`	`LON` instance constructed from the sampling trace.

Source code in src/lonkit/continuous/sampling.py

def sample_to_lon(
    self,
    sampler_result: BasinHoppingResult,
    lon_config: LONConfig | None = None,
) -> LON:
    """
    Construct a LON from a `BasinHoppingResult`.

    Convenience wrapper that passes the trace data from a sampling result
    to `LON.from_trace_data()`. Equivalent to calling
    `LON.from_trace_data(sampler_result.trace_df, config=lon_config)`.

    Args:
        sampler_result: Result returned by `sample()`.
        lon_config: LON construction configuration. If `None`, uses default
            `LONConfig`. Default: `None`.

    Returns:
        `LON` instance constructed from the sampling trace.
    """

    trace_df = sampler_result.trace_df

    if trace_df.empty:
        return LON()

    _lon_config = replace(lon_config) if lon_config is not None else LONConfig()

    if _lon_config.eq_atol is None:
        p = self.config.fitness_precision
        if p is not None and p >= 0:
            _lon_config.eq_atol = 10 ** -(p + 1)

    return LON.from_trace_data(trace_df, config=_lon_config)

BasinHoppingResult `dataclass` ¶

Result of a Basin-Hopping sampling run.

Attributes:

Name	Type	Description
`trace_df`	`DataFrame`	DataFrame with columns `[run, fit1, node1, fit2, node2]` representing transitions between local optima.
`raw_records`	`list[dict]`	List of dicts with detailed iteration data, including all perturbation attempts (both accepted and rejected). Each dict has keys `run`, `iteration`, `current_x`, `current_f`, `new_x`, `new_f`, and `accepted`.
`nfev`	`int`	Total number of objective function evaluations across all runs.

Source code in src/lonkit/continuous/sampling.py

@dataclass
class BasinHoppingResult:
    """
    Result of a Basin-Hopping sampling run.

    Attributes:
        trace_df: DataFrame with columns `[run, fit1, node1, fit2, node2]` representing
            transitions between local optima.
        raw_records: List of dicts with detailed iteration data, including all
            perturbation attempts (both accepted and rejected). Each dict has keys
            `run`, `iteration`, `current_x`, `current_f`, `new_x`, `new_f`, and `accepted`.
        nfev: Total number of objective function evaluations across all runs.
    """

    trace_df: pd.DataFrame
    raw_records: list[dict]
    nfev: int

Step Size Estimation¶

StepSizeEstimatorConfig `dataclass` ¶

Configuration for step size estimation.

Attributes:

Name	Type	Description
`n_samples`	`int`	Number of random initial points to evaluate. Default: `100`.
`n_perturbations`	`int`	Number of perturbations per sample point. Default: `30`.
`target_escape_rate`	`float`	Target escape rate to find (0.5 = 50% of perturbations escape). Default: `0.5`.
`search_precision`	`int`	Decimal digits of precision for step size search. The algorithm refines by dividing the increment by 10 each iteration, so `search_precision=4` means 4 refinement rounds yielding resolution 0.0001. Default: `4`.
`coordinate_precision`	`int \| None`	Decimal precision for coordinate rounding and hashing. Solutions rounded to this precision are considered identical. Use `None` for full double precision (no rounding). Default: `4`.
`minimizer_method`	`str \| Callable \| None`	Minimization method passed to `scipy.optimize.minimize`. Can be a string or a callable implementing a custom solver. See `scipy.optimize.minimize <https://docs.scipy.org/doc/scipy/reference/generated/scipy.optimize.minimize.html>`_ for the full list of supported methods and their options. Default: `"L-BFGS-B"`.
`minimizer_options`	`dict \| None`	Solver-specific options passed as the `options` argument to `scipy.optimize.minimize`. The available keys depend on the chosen `minimizer_method`. Use `None` to rely on scipy's defaults. Default: `None`.
`bounded`	`bool`	Whether to enforce domain bounds during perturbation. Default: `True`.
`seed`	`int \| None`	Random seed for reproducibility. Default: `None`.

Source code in src/lonkit/continuous/step_size.py

@dataclass
class StepSizeEstimatorConfig:
    """
    Configuration for step size estimation.

    Attributes:
        n_samples: Number of random initial points to evaluate. Default: `100`.
        n_perturbations: Number of perturbations per sample point. Default: `30`.
        target_escape_rate: Target escape rate to find (0.5 = 50% of perturbations escape). Default: `0.5`.
        search_precision: Decimal digits of precision for step size search.
            The algorithm refines by dividing the increment by 10 each iteration,
            so ``search_precision=4`` means 4 refinement rounds yielding resolution 0.0001. Default: `4`.
        coordinate_precision: Decimal precision for coordinate rounding and hashing.
            Solutions rounded to this precision are considered identical.
            Use `None` for full double precision (no rounding). Default: `4`.
        minimizer_method: Minimization method passed to ``scipy.optimize.minimize``. Can be a
            string or a callable implementing a custom solver.
            See `scipy.optimize.minimize <https://docs.scipy.org/doc/scipy/reference/generated/scipy.optimize.minimize.html>`_
            for the full list of supported methods and their options. Default: `"L-BFGS-B"`.
        minimizer_options: Solver-specific options passed as the ``options`` argument to
            ``scipy.optimize.minimize``. The available keys depend on the chosen
            ``minimizer_method``. Use ``None`` to rely on scipy's defaults.
            Default: `None`.
        bounded: Whether to enforce domain bounds during perturbation. Default: `True`.
        seed: Random seed for reproducibility. Default: `None`.
    """

    n_samples: int = 100
    n_perturbations: int = 30
    target_escape_rate: float = 0.5
    search_precision: int = 4
    coordinate_precision: int | None = 4
    minimizer_method: str | Callable | None = "L-BFGS-B"
    minimizer_options: dict | None = None
    bounded: bool = True
    seed: int | None = None

    def __post_init__(self) -> None:
        if self.n_samples <= 0:
            raise ValueError("n_samples must be greater than 0.")
        if self.n_perturbations <= 0:
            raise ValueError("n_perturbations must be greater than 0.")
        if not (0 <= self.target_escape_rate <= 1):
            raise ValueError("target_escape_rate must be between 0 and 1.")
        if self.search_precision < 1:
            raise ValueError("search_precision must be at least 1.")

StepSizeEstimator ¶

Estimates the optimal percentage step size for basin-hopping sampling.

The optimal step size is defined as the one that produces an escape rate closest to a target X (default 0.5), meaning ~ X * 100% of perturbations lead to a different local optimum.

The search uses a decimal refinement approach, progressively narrowing the step size to the configured precision.

Computational cost

_compute_escape_rate is called once per tested step size. Each call runs n_samples baseline minimizations and n_samples * n_perturbations perturbed minimizations (defaults: 100 + 3000 minimizations per step size). Since multiple step sizes are evaluated during refinement (search_precision dependent), total minimizations can become large for expensive objective functions. For expensive objectives, start by reducing n_samples and/or n_perturbations, then increase them once a reasonable step-size range is identified.

Example

import numpy as np estimator = StepSizeEstimator() result = estimator.estimate(problem, [(-5, 5)] * 2) print(f"Step size: {result.step_size}, escape rate: {result.escape_rate:.3f}")

Source code in src/lonkit/continuous/step_size.py

class StepSizeEstimator:
    """
    Estimates the optimal percentage step size for basin-hopping sampling.

    The optimal step size is defined as the one that produces an escape rate
    closest to a target ``X`` (default 0.5), meaning ~ ``X * 100%`` of perturbations lead to
    a different local optimum.

    The search uses a decimal refinement approach, progressively narrowing
    the step size to the configured precision.

    Computational cost:
        `_compute_escape_rate` is called once per tested step size. Each call runs
        `n_samples` baseline minimizations and `n_samples * n_perturbations`
        perturbed minimizations (defaults: 100 + 3000 minimizations per step size).
        Since multiple step sizes are evaluated during refinement
        (`search_precision` dependent), total minimizations can become large for
        expensive objective functions. For expensive objectives, start by reducing
        `n_samples` and/or `n_perturbations`, then increase them once a reasonable
        step-size range is identified.

    Example:
        >>> import numpy as np
        >>> estimator = StepSizeEstimator()
        >>> result = estimator.estimate(problem, [(-5, 5)] * 2)
        >>> print(f"Step size: {result.step_size}, escape rate: {result.escape_rate:.3f}")
    """

    def __init__(self, config: StepSizeEstimatorConfig | None = None):
        self.config = config or StepSizeEstimatorConfig()

    def _make_sampler(self) -> BasinHoppingSampler:
        sampler_config = BasinHoppingSamplerConfig(
            coordinate_precision=self.config.coordinate_precision,
            bounded=self.config.bounded,
            minimizer_method=self.config.minimizer_method,
            minimizer_options=self.config.minimizer_options,
            step_mode="percentage",
            seed=self.config.seed,
        )
        return BasinHoppingSampler(sampler_config)

    def _compute_escape_rate(
        self,
        func: Callable[[np.ndarray], float],
        domain_array: np.ndarray,
        step_size: float,
        sampler: BasinHoppingSampler,
    ) -> float:
        """
        Compute the average escape rate for a given step size.

        Cost per call: `n_samples` + `n_samples * n_perturbations` minimizations.
        """
        bounds_array = domain_array if self.config.bounded else None
        p = step_size * np.abs(domain_array[:, 1] - domain_array[:, 0])
        rng = np.random.default_rng(self.config.seed)

        escape_rates = []

        for _ in range(self.config.n_samples):
            x0 = rng.uniform(domain_array[:, 0], domain_array[:, 1])
            try:
                res = minimize(
                    func,
                    x0,
                    method=self.config.minimizer_method,
                    options=self.config.minimizer_options,
                    bounds=bounds_array if self.config.bounded else None,
                )
            except ValueError as e:
                warnings.warn(
                    f"Initial minimize failed with ValueError: {e}. "
                    f"Starting point: {x0}. Skipping run.",
                    stacklevel=3,
                )
                continue
            optimum = res.x
            optimum_rounded = sampler._round_value(optimum, self.config.coordinate_precision)
            optimum_hash = sampler._hash_solution(optimum_rounded)

            escapes = 0
            for _ in range(self.config.n_perturbations):
                x_perturbed = sampler._perturbation(optimum, p, rng, bounds_array)
                try:
                    res_perturbed = minimize(
                        func,
                        x_perturbed,
                        method=self.config.minimizer_method,
                        options=self.config.minimizer_options,
                        bounds=bounds_array,
                    )
                except ValueError as e:
                    # L-BFGS-B can produce internal iterates that slightly
                    # violate bounds, causing approx_derivative to fail.
                    # Skip this perturbation and try the next one.
                    warnings.warn(
                        f"Minimize after perturbation "
                        f"failed with ValueError: {e}. "
                        f"Perturbed point: {x_perturbed}. Skipping perturbation.",
                        stacklevel=3,
                    )
                    continue
                new_hash = sampler._hash_solution(res_perturbed.x)
                if new_hash != optimum_hash:
                    escapes += 1

            escape_rates.append(escapes / self.config.n_perturbations)

        return float(np.mean(escape_rates))

    def estimate(
        self,
        func: Callable[[np.ndarray], float],
        domain: list[tuple[float, float]],
    ) -> StepSizeResult:
        """
        Estimate the optimal step size for basin-hopping sampling.

        Args:
            func: Objective function to minimize (f: R^n -> R).
            domain: List of (lower, upper) bounds per dimension.

        Returns:
            StepSizeResult with the estimated step size, achieved escape rate, and error.
        """
        domain_array = np.array(domain)
        sampler = self._make_sampler()

        step = 0.1
        increment = 0.1
        target = self.config.target_escape_rate

        best_lower: tuple[float, float] | None = None  # (step, rate)
        best_upper: tuple[float, float] | None = None  # (step, rate)
        last_tested: tuple[float, float] | None = None

        for _ in range(self.config.search_precision):
            while step <= 1.0 + 1e-12:  # epsilon to ensure step=1.0 is tested despite float drift
                rate = self._compute_escape_rate(func, domain_array, step, sampler)
                last_tested = (step, rate)

                if rate < target:
                    if best_lower is None or abs(rate - target) < abs(best_lower[1] - target):
                        best_lower = (step, rate)
                    step += increment
                else:
                    if best_upper is None or abs(rate - target) < abs(best_upper[1] - target):
                        best_upper = (step, rate)
                    # Refine: reduce increment and resume from lower bound
                    increment /= 10
                    if best_lower is not None:
                        step = best_lower[0] + increment
                    else:
                        step = increment
                    break
            else:
                # Reached step > 1.0 without finding upper bound
                break

        # Select the candidate closest to target
        candidates = [c for c in (best_lower, best_upper, last_tested) if c is not None]
        best_step, best_rate = min(candidates, key=lambda c: abs(c[1] - target))

        return StepSizeResult(
            step_size=round(best_step, self.config.search_precision),
            escape_rate=best_rate,
            error=abs(best_rate - target),
        )

estimate ¶

estimate(
    func: Callable[[ndarray], float],
    domain: list[tuple[float, float]],
) -> StepSizeResult

Estimate the optimal step size for basin-hopping sampling.

Parameters:

Name	Type	Description	Default
`func`	`Callable[[ndarray], float]`	Objective function to minimize (f: R^n -> R).	required
`domain`	`list[tuple[float, float]]`	List of (lower, upper) bounds per dimension.	required

Returns:

Type	Description
`StepSizeResult`	StepSizeResult with the estimated step size, achieved escape rate, and error.

Source code in src/lonkit/continuous/step_size.py

def estimate(
    self,
    func: Callable[[np.ndarray], float],
    domain: list[tuple[float, float]],
) -> StepSizeResult:
    """
    Estimate the optimal step size for basin-hopping sampling.

    Args:
        func: Objective function to minimize (f: R^n -> R).
        domain: List of (lower, upper) bounds per dimension.

    Returns:
        StepSizeResult with the estimated step size, achieved escape rate, and error.
    """
    domain_array = np.array(domain)
    sampler = self._make_sampler()

    step = 0.1
    increment = 0.1
    target = self.config.target_escape_rate

    best_lower: tuple[float, float] | None = None  # (step, rate)
    best_upper: tuple[float, float] | None = None  # (step, rate)
    last_tested: tuple[float, float] | None = None

    for _ in range(self.config.search_precision):
        while step <= 1.0 + 1e-12:  # epsilon to ensure step=1.0 is tested despite float drift
            rate = self._compute_escape_rate(func, domain_array, step, sampler)
            last_tested = (step, rate)

            if rate < target:
                if best_lower is None or abs(rate - target) < abs(best_lower[1] - target):
                    best_lower = (step, rate)
                step += increment
            else:
                if best_upper is None or abs(rate - target) < abs(best_upper[1] - target):
                    best_upper = (step, rate)
                # Refine: reduce increment and resume from lower bound
                increment /= 10
                if best_lower is not None:
                    step = best_lower[0] + increment
                else:
                    step = increment
                break
        else:
            # Reached step > 1.0 without finding upper bound
            break

    # Select the candidate closest to target
    candidates = [c for c in (best_lower, best_upper, last_tested) if c is not None]
    best_step, best_rate = min(candidates, key=lambda c: abs(c[1] - target))

    return StepSizeResult(
        step_size=round(best_step, self.config.search_precision),
        escape_rate=best_rate,
        error=abs(best_rate - target),
    )

StepSizeResult `dataclass` ¶

Result of step size estimation.

Attributes:

Name	Type	Description
`step_size`	`float`	Estimated optimal step size (percentage of domain range).
`escape_rate`	`float`	Achieved escape rate at this step size.
`error`	`float`	Absolute difference between achieved and target escape rate.

Source code in src/lonkit/continuous/step_size.py

@dataclass(frozen=True)
class StepSizeResult:
    """
    Result of step size estimation.

    Attributes:
        step_size: Estimated optimal step size (percentage of domain range).
        escape_rate: Achieved escape rate at this step size.
        error: Absolute difference between achieved and target escape rate.
    """

    step_size: float
    escape_rate: float
    error: float

Continuous Sampling Module¶

compute_lon ¶

BasinHoppingSamplerConfig dataclass ¶

BasinHoppingSampler ¶

sample ¶

sample_to_lon ¶

BasinHoppingResult dataclass ¶

Step Size Estimation¶

StepSizeEstimatorConfig dataclass ¶

StepSizeEstimator ¶

estimate ¶

StepSizeResult dataclass ¶

BasinHoppingSamplerConfig `dataclass` ¶

BasinHoppingResult `dataclass` ¶

StepSizeEstimatorConfig `dataclass` ¶

StepSizeResult `dataclass` ¶