Skip to content

Benchmark

BenchmarkConfig(n_hubs, n_nodes_range, n_clients_range, methods, random_nodes=True, random_n_clients_route=False, alpha=0.7, min_budget_factor=0.2, max_budget_factor=5.0, time_limit=3600, mip_gap=0.0001, int_feas_tol=1e-09, benchmark_type='n_nodes', file_name='benchmark_results', output_dir='benchmark_results', seed_dict=(lambda: {run: run for run in (range(1, 11))})(), n_runs=10, scaling=True, scaling_factor=1000) dataclass

Benchmark job configuration.

Holds everything needed to describe a single benchmark queue entry: what ranges to sweep, which methods to run, and how results are saved.

BenchmarkResult(n_nodes, n_hubs, alpha, method, total_time, solving_time, lagrange_time, optimal, gap, node_count, n_clients, n_clients_route, random_n_clients_route, min_budget_factor, max_budget_factor, run_number, seed, sanctioned=False) dataclass

Stores the result of a single benchmark solve.

Captures the configuration parameters (nodes, hubs, clients, seed) together with the solution metrics returned by a specific model/solver run (solving time, optimality gap, node count, etc.).

from_solution(solution, config, n_nodes, n_clients, run, method) classmethod

Construct from a [BaseModelSolution][bilevelpy.solution.core.BaseModelSolution] and benchmark configuration.

Parameters:

Name Type Description Default
solution BaseModelSolution

The solution returned by the solver.

 required
config BenchmarkConfig

The benchmark configuration.

 required
n_nodes int

Number of nodes in the instance.

 required
n_clients int

Number of clients per route.

 required
run int

Run number (1-indexed, for seed lookup).

 required
method str

Method string key (e.g. "pc_hlp").

 required
Source code in src/oracle_paper/benchmark/result.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@classmethod
def from_solution(
    cls,
    solution: BaseModelSolution,
    config: BenchmarkConfig,
    n_nodes: int,
    n_clients: int,
    run: int,
    method: str,
) -> "BenchmarkResult":
    """Construct from a
    [`BaseModelSolution`][bilevelpy.solution.core.BaseModelSolution]
    and benchmark configuration.

    Args:
        solution: The solution returned by the solver.
        config: The benchmark configuration.
        n_nodes: Number of nodes in the instance.
        n_clients: Number of clients per route.
        run: Run number (1-indexed, for seed lookup).
        method: Method string key (e.g. ``"pc_hlp"``).
    """
    meta = solution.solution_metadata
    extra = meta.extra if hasattr(meta, "extra") else {}

    lagrange_time = extra.get("lagrange_time", 0.0) or 0.0
    total_time = float(meta.solving_time) + float(lagrange_time)

    return cls(
        n_nodes=n_nodes,
        n_hubs=config.n_hubs,
        alpha=config.alpha,
        method=method,
        total_time=total_time,
        solving_time=float(meta.solving_time),
        lagrange_time=float(lagrange_time),
        optimal=meta.is_optimal,
        gap=float(meta.mip_gap),
        node_count=int(meta.node_count),
        n_clients=extra.get("n_clients", 0),
        n_clients_route=n_clients,
        random_n_clients_route=config.random_n_clients_route,
        min_budget_factor=config.min_budget_factor,
        max_budget_factor=config.max_budget_factor,
        run_number=run,
        seed=config.seed_dict[run],
    )

sanctioned_method(config, method, n_nodes, n_clients, run) classmethod

Create a placeholder result for a sanctioned (skipped) method.

Source code in src/oracle_paper/benchmark/result.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@classmethod
def sanctioned_method(
    cls,
    config: BenchmarkConfig,
    method: str,
    n_nodes: int,
    n_clients: int,
    run: int,
) -> "BenchmarkResult":
    """Create a placeholder result for a sanctioned (skipped) method."""
    return cls(
        n_nodes=n_nodes,
        n_hubs=config.n_hubs,
        alpha=config.alpha,
        method=method,
        total_time=None,
        solving_time=None,
        lagrange_time=None,
        optimal=None,
        gap=None,
        node_count=None,
        n_clients=None,
        n_clients_route=n_clients,
        random_n_clients_route=config.random_n_clients_route,
        min_budget_factor=config.min_budget_factor,
        max_budget_factor=config.max_budget_factor,
        run_number=run,
        seed=config.seed_dict[run],
        sanctioned=True,
    )

to_dict()

Convert to a plain dictionary.

Source code in src/oracle_paper/benchmark/result.py 
48
49
50
def to_dict(self) -> Dict[str, Any]:
    """Convert to a plain dictionary."""
    return asdict(self)

BenchmarkResults(save=False)

Manages a collection of BenchmarkResult instances.

Provides utilities to add results, export to CSV, filter, and iterate.

Source code in src/oracle_paper/benchmark/results.py 
20
21
def __init__(self, save: bool = False):
    self._results: List[BenchmarkResult] = []

add(result)

Add a result to the collection.

Source code in src/oracle_paper/benchmark/results.py 
23
24
25
def add(self, result: BenchmarkResult) -> None:
    """Add a result to the collection."""
    self._results.append(result)

filter_by_method(method)

Filter stored results by method name.

Source code in src/oracle_paper/benchmark/results.py 
39
40
41
def filter_by_method(self, method: str) -> List[BenchmarkResult]:
    """Filter stored results by method name."""
    return [r for r in self._results if r.method == method]

save_to_csv(path)

Save results to a CSV file.

Source code in src/oracle_paper/benchmark/results.py 
35
36
37
def save_to_csv(self, path: str) -> None:
    """Save results to a CSV file."""
    self.to_dataframe().to_csv(path, index=False)

to_dataframe()

Convert results to a pandas DataFrame.

Source code in src/oracle_paper/benchmark/results.py 
31
32
33
def to_dataframe(self) -> pd.DataFrame:
    """Convert results to a pandas DataFrame."""
    return pd.DataFrame(self.to_dicts())

to_dicts()

Convert all results to a list of dictionaries.

Source code in src/oracle_paper/benchmark/results.py 
27
28
29
def to_dicts(self) -> List[dict]:
    """Convert all results to a list of dictionaries."""
    return [r.to_dict() for r in self._results]

BenchmarkResultsSaver(bench_config)

Saves raw benchmark results to timestamped folders.

File paths are constructed from benchmark configuration and the current date/time.

Source code in src/oracle_paper/benchmark/results_saver.py 
21
22
23
24
25
26
27
28
29
30
def __init__(self, bench_config: BenchmarkConfig) -> None:
    self.bench_config = bench_config
    self.file_name = bench_config.file_name
    self.file_path = self._build_file_path()

    os.makedirs(self.file_path, exist_ok=True)
    self._logger = self._init_logger()

    # Full base path (no extension)
    self.file_name = os.path.join(self.file_path, self.file_name)

save_benchmark_raw_results(benchmark_results)

Save raw benchmark results to CSV.

Parameters:

Name Type Description Default
benchmark_results BenchmarkResults

The results collection to persist.

 required
Source code in src/oracle_paper/benchmark/results_saver.py 
82
83
84
85
86
87
88
89
def save_benchmark_raw_results(self, benchmark_results: BenchmarkResults) -> None:
    """Save raw benchmark results to CSV.

    Args:
        benchmark_results: The results collection to persist.
    """
    raw_csv = self.file_name + "_raw.csv"
    benchmark_results.save_to_csv(raw_csv)

BenchmarkRunner(config)

Runs benchmark experiments from a BenchmarkConfig.

For each scenario (varying n_nodes or n_clients), it calls PaperModelProvider.build_and_solve for every method × run combination, collects BenchmarkResult entries, and streams raw results to disk after each solve.

Supports optional progress and result callbacks for live UIs.

Source code in src/oracle_paper/benchmark/runner.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(self, config: BenchmarkConfig) -> None:
    self.config = config
    self.provider = PaperModelProvider(config=config)

    self.benchmark_saver = BenchmarkResultsSaver(config)
    self._results = BenchmarkResults()

    # Callbacks
    self.callbacks: List[Callable[[BenchmarkResults], None]] = []
    self.progress_callback: Callable[[int, int, str], None] | None = None
    self.total_steps = 0
    self.current_step = 0

    # Sanctioning state (per method string key)
    self.sanctions_dict: Dict[str, int] = dict.fromkeys(config.methods, 0)
    self.is_sanctioned: Dict[str, bool] = dict.fromkeys(config.methods, False)

add_callback(callback_fn)

Register a callback invoked after each result is added.

Source code in src/oracle_paper/benchmark/runner.py 
54
55
56
def add_callback(self, callback_fn: Callable[[BenchmarkResults], None]) -> None:
    """Register a callback invoked after each result is added."""
    self.callbacks.append(callback_fn)

run()

Execute the full benchmark grid.

Source code in src/oracle_paper/benchmark/runner.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def run(self) -> None:
    """Execute the full benchmark grid."""
    logger = self.benchmark_saver.logger
    logger.info(f"Running benchmark: {self.config.benchmark_type}")

    scenarios = self._build_scenarios()
    self.total_steps = (
        len(scenarios) * self.config.n_runs * len(self.config.methods)
    )
    self.current_step = 0

    for scenario in scenarios:
        n_nodes = scenario["n_nodes"]
        n_clients = scenario["n_clients"]
        logger.info(f"n_nodes={n_nodes}, n_clients={n_clients}")

        for run in range(1, self.config.n_runs + 1):
            seed = self.config.seed_dict[run]
            run_idx = run - 1  # 0-indexed for the provider
            logger.info(f"\tRun={run}")

            for method_str in self.config.methods:
                self.current_step += 1

                if self.progress_callback:
                    info = (
                        f"Nodes: {n_nodes} | Clients: {n_clients} | "
                        f"Run: {run}/{self.config.n_runs} | Method: {method_str}"
                    )
                    self.progress_callback(
                        self.current_step, self.total_steps, info
                    )

                # --- Sanctioning logic ---
                if run == 1:
                    if self.sanctions_dict[method_str] >= self.config.n_runs:
                        self.is_sanctioned[method_str] = True
                        logger.warning(
                            f"\t\tMethod {method_str} has been sanctioned."
                        )
                    else:
                        self.sanctions_dict[method_str] = 0

                if self.is_sanctioned[method_str]:
                    result = BenchmarkResult.sanctioned_method(
                        config=self.config,
                        method=method_str,
                        n_nodes=n_nodes,
                        n_clients=n_clients,
                        run=run,
                    )
                    self._results.add(result)
                    self.benchmark_saver.save_benchmark_raw_results(self._results)
                    self._notify_callbacks()
                    logger.warning(
                        f"\t\tSanctioned method {method_str} skipped."
                    )
                    continue

                # --- Build & solve ---
                model_meta = METHOD_MAP.get(method_str)
                if model_meta is None:
                    logger.warning(
                        f"\t\tUnknown method '{method_str}', skipping."
                    )
                    result = BenchmarkResult.sanctioned_method(
                        config=self.config,
                        method=method_str,
                        n_nodes=n_nodes,
                        n_clients=n_clients,
                        run=run,
                    )
                    self._results.add(result)
                    self.benchmark_saver.save_benchmark_raw_results(self._results)
                    self._notify_callbacks()
                    continue

                try:
                    solution = self.provider.build_and_solve(
                        model_name=model_meta,
                        scenario=scenario,
                        run_idx=run_idx,
                        seed=seed,
                    )
                except Exception:
                    logger.exception(
                        "Unexpected benchmark failure for %s",
                        method_str,
                    )
                    raise

                # --- Collect result ---
                result = BenchmarkResult.from_solution(
                    solution=solution,
                    config=self.config,
                    n_nodes=n_nodes,
                    n_clients=n_clients,
                    run=run,
                    method=method_str,
                )

                if not result.optimal:
                    self.sanctions_dict[method_str] += 1
                    logger.warning(
                        f"\t\tSolution not optimal for {method_str}, "
                        f"sanction counter: {self.sanctions_dict[method_str]}"
                    )
                else:
                    logger.info(
                        f"\t\tOptimal solution for {method_str}, "
                        f"solving time: {result.solving_time:.2f}s"
                    )

                self._results.add(result)
                self.benchmark_saver.save_benchmark_raw_results(self._results)
                self._notify_callbacks()

                # Dispose Gurobi model to free memory
                try:
                    solution.dispose()
                except Exception:
                    pass

set_progress_callback(callback)

Register a progress callback: fn(current, total, status_text).

Source code in src/oracle_paper/benchmark/runner.py 
58
59
60
61
62
def set_progress_callback(
    self, callback: Callable[[int, int, str], None]
) -> None:
    """Register a progress callback: ``fn(current, total, status_text)``."""
    self.progress_callback = callback

PaperModelProvider(config)

Bases: ModelProvider

Builds datasets and instantiates models for benchmark runs.

Handles the full pipeline for each of the four models studied in the paper. For a given scenario (nodes, clients, hubs), it:

  1. Builds the dataset via the standard pipeline (CAB loader → node selection → cost scaling → client generation → client ranking).
  2. Runs the appropriate calculators (Lagrange and/or recursive Lagrange) depending on the model.
  3. Instantiates the model with the configured parameters.
  4. Solves and returns the solution with metadata attached.
Source code in src/oracle_paper/benchmark/model_provider.py 
39
40
def __init__(self, config: BenchmarkConfig):
    self._config = config

build_and_solve(model_name, scenario, run_idx, seed)

Build dataset, instantiate model, solve, and return the solution.

Parameters:

Name Type Description Default
model_name ModelMetaData

Which model to run (from OraclePaperModelNames).

 required
scenario Dict[str, Any]

Dict with n_nodes, n_clients, n_hubs, and alpha.

 required
run_idx int

Zero-based run index (for dataset seed offset).

 required
seed int

Random seed for reproducibility.

 required

Returns:

Type Description
BaseModelSolution

The solution produced by
BaseModelSolution

[ModelSolver][bilevelpy.solver.ModelSolver].

Raises:

Type Description
ValueError

If model_name is not one of the four known models.
Source code in src/oracle_paper/benchmark/model_provider.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def build_and_solve(self,
                    model_name: ModelMetaData,
                    scenario: Dict[str, Any],
                    run_idx: int,
                    seed: int) -> BaseModelSolution:
    """Build dataset, instantiate model, solve, and return the solution.

    Args:
        model_name: Which model to run (from
            [`OraclePaperModelNames`][oracle_paper.core.names.OraclePaperModelNames]).
        scenario: Dict with ``n_nodes``, ``n_clients``, ``n_hubs``,
            and ``alpha``.
        run_idx: Zero-based run index (for dataset seed offset).
        seed: Random seed for reproducibility.

    Returns:
        The solution produced by
        [`ModelSolver`][bilevelpy.solver.ModelSolver].

    Raises:
        ValueError: If ``model_name`` is not one of the four known models.
    """

    n_nodes = scenario.get("n_nodes")
    n_clients = scenario.get("n_clients")
    n_hubs = scenario.get("n_hubs")
    alpha = scenario.get("alpha")

    builder = (
        DatasetBuilder()
        .pipe(HLPLoader(Dataset.CAB100))
        .pipe(
            HLPNodeSelector(
                n_nodes=n_nodes,
                random_nodes=self._config.random_nodes,
                seed=seed,
            )
        )
    )

    if self._config.scaling:
        builder.pipe(
            HLPCostScaling(
                scaling_factor=self._config.scaling_factor
            )
        )

    builder.pipe(
        BilevelClientGenerator(
            clients_per_route=n_clients,
            random_count=self._config.random_n_clients_route,
            min_budget_factor=self._config.min_budget_factor,
            max_budget_factor=self._config.max_budget_factor,
            possible_weights=list(range(1, 21)),
            seed=seed,
        )
    )

    builder.pipe(LinearClientRanker())

    processors_metadata = {}

    if model_name == OraclePaperModelNames.PC_HLP:
        lagrange_calc = LagrangeCalculator()
        recursive_lagrange_calc = RecursiveLagrangeCalculator()

        dataset = (builder
                   .pipe(lagrange_calc)
                   .pipe(recursive_lagrange_calc)
                   .build())

        total_lagrange_time =  (lagrange_calc.get_all_metrics()["lagrange_time"]
            + recursive_lagrange_calc.get_all_metrics()["recursive_lagrange_time"])

        processors_metadata.update({
            "lagrange_time": total_lagrange_time,})

        model = PC_HLP(n_hubs=n_hubs, alpha=alpha, data=dataset)

    elif model_name == OraclePaperModelNames.PPC_HLP:
        lagrange_calc = LagrangeCalculator()

        dataset = (builder
                   .pipe(lagrange_calc)
                   .build())


        processors_metadata.update({
            "lagrange_time": lagrange_calc.get_all_metrics()["lagrange_time"],
        })

        model = PPC_HLP(n_hubs=n_hubs, alpha=alpha, data=dataset)

    elif model_name in (OraclePaperModelNames.PS_HLP, OraclePaperModelNames.PS_BHLP):
        dataset = builder.build()
        model_cls = PS_HLP if model_name == OraclePaperModelNames.PS_HLP else PS_BHLP
        model = model_cls(n_hubs=n_hubs, alpha=alpha, data=dataset)

    else:
        raise ValueError(f"Unknown method requested: {model_name}")



    solution = ModelSolver(model=model,
                           use_max_threads=False,
                           time_limit=self._config.time_limit).solve()

    solution.solution_metadata.extra.update(processors_metadata)

    total_n_clients = len(dataset[BilevelDataCol.CLIENT_KEY])
    solution.solution_metadata.extra["n_clients"] = total_n_clients

    print(f"Solution Metadata of {model_name}:\n{solution.solution_metadata}")

    return solution

main()

Launch the Streamlit benchmark UI.

Source code in src/oracle_paper/benchmark/__init__.py 
22
23
24
25
def main() -> None:
    """Launch the Streamlit benchmark UI."""
    app = Path(__file__).resolve().parents[3] / "reproduce" / "benchmarking_tool.py"
    subprocess.run([sys.executable, "-m", "streamlit", "run", str(app)])