Data

Data pipeline components for bilevel hub location datasets.

Generators create synthetic clients and budgets, processors rank and index clients, and calculators compute Lagrange multipliers.

`BilevelClientGenerator(clients_per_route, random_count=False, min_budget_factor=1.2, max_budget_factor=1.8, possible_weights=None, seed=42)` ¶

Bases: EntityProcessor

Generate synthetic bilevel clients on each route.

For every route $(i,j)$ with $i \neq j$, generates a random number of clients, each with a weight $a_{ij}^z$ (sampled from possible_weights) and a budget $b_{ij}^z = c_{ij} \cdot a_{ij}^z \cdot \text{factor}$, where the factor is uniformly sampled from $[\text{min\_budget\_factor}, \text{max\_budget\_factor}]$.

Parameters:

Name	Type	Description	Default
`clients_per_route`	`int`	Number of clients per route (or max if `random_count` is `True`).	required
`random_count`	`bool`	If `True`, each route gets a random number of clients between 1 and `clients_per_route`.	`False`
`min_budget_factor`	`float`	Minimum multiplier for budget generation.	`1.2`
`max_budget_factor`	`float`	Maximum multiplier for budget generation.	`1.8`
`possible_weights`	`List[float]`	List of possible client weights to sample from.	`None`
`seed`	`int`	Random seed for reproducibility.	`42`

Source code in src/oracle_paper/data/generator/bilevel_client_generator.py

def __init__(
    self,
    clients_per_route: int,
    random_count: bool = False,
    min_budget_factor: float = 1.2,
    max_budget_factor: float = 1.8,
    possible_weights: List[float] = None,
    seed: int = 42,
):
    self.clients_per_route = clients_per_route
    self.random_count = random_count
    self.min_budget_factor = min_budget_factor
    self.max_budget_factor = max_budget_factor
    self.possible_weights = possible_weights or list(range(1,21))
    self.rng = random.Random(seed)

`LagrangeCalculator()` ¶

Bases: TrackableProcessor

Compute Lagrange multipliers $\lambda_{ij}^z$ for the PPC-HLP model.

For each route $(i,j)$, the calculator sorts clients by their budget-to-weight ratio and computes the Lagrange multiplier sequence using the recursive formula:

\[\lambda_k = b_k + \sum_{t=0}^{k-1} a_t \cdot \left(\frac{b_k}{a_k} - \frac{b_{k-1}}{a_{k-1}}\right)\]

The result is stored in BilevelDataCol.LAGRANGE.

Uses the track_metric decorator to measure computation time automatically.

Source code in src/oracle_paper/data/calculator/base.py

def __init__(self):
	"""Initialize the processor with metadata storage."""
	super().__init__()
	self._processor_metadata: dict[str, Any] = {}

`calculate_lagrange(dict_a, dict_b)` `staticmethod` ¶

Compute the Lagrange multiplier sequence for one route's clients.

$$\lambda_k = b_k + \sum_{t=0}^{k-1} a_t \cdot \Delta_k$$ where $\Delta_k = b_k/a_k - b_{k-1}/a_{k-1}$.

Parameters:

Name	Type	Description	Default
`dict_a`	`dict[int, float]`	Client weights $\{z: a_{ij}^z\}$ sorted by $z$.	required
`dict_b`	`dict[int, float]`	Client budgets $\{z: b_{ij}^z\}$ sorted by $z$.	required

Returns:

Type	Description
`dict[int, float]`	Mapping $\{z: \lambda_{ij}^z\}$ of Lagrange multipliers.

Source code in src/oracle_paper/data/calculator/lagrange.py

@staticmethod
def calculate_lagrange(
    dict_a: dict[int, float], dict_b: dict[int, float]
) -> dict[int, float]:
    r"""Compute the Lagrange multiplier sequence for one route's clients.

    $$\lambda_k = b_k + \sum_{t=0}^{k-1} a_t \cdot \Delta_k$$
    where $\Delta_k = b_k/a_k - b_{k-1}/a_{k-1}$.

    Args:
        dict_a: Client weights $\{z: a_{ij}^z\}$ sorted by $z$.
        dict_b: Client budgets $\{z: b_{ij}^z\}$ sorted by $z$.

    Returns:
        Mapping $\{z: \lambda_{ij}^z\}$ of Lagrange multipliers.
    """
    curr_sum = 0.0
    lagrange = {}

    sorted_z = sorted(dict_a.keys())

    for k in sorted_z:
        if k == 0:
            curr_lagrange = dict_b[k]
        else:
            curr_lagrange = dict_b[k] + curr_sum * (
                    (dict_b[k] / dict_a[k]) - (dict_b[k - 1] / dict_a[k - 1])
            )
        curr_sum += dict_a[k]
        lagrange[k] = curr_lagrange
    return lagrange

`process(dataset)` ¶

Compute Lagrange multipliers and add them to the dataset.

Parameters:

Name	Type	Description	Default
`dataset`	`MultiEntityDataset`	Dataset with client weights and budgets (modified in-place).	required

Raises:

Type	Description
`AttributeError`	If required entities are missing.

Source code in src/oracle_paper/data/calculator/lagrange.py

@track_metric("lagrange_time")
def process(self, dataset: MultiEntityDataset) -> None:
    """Compute Lagrange multipliers and add them to the dataset.

    Args:
        dataset: Dataset with client weights and budgets
            (modified in-place).

    Raises:
        AttributeError: If required entities are missing.
    """

    if BilevelDataCol.TRANSPORT_WEIGHT_CLIENT not in dataset:
        raise AttributeError(f"{BilevelDataCol.TRANSPORT_WEIGHT_CLIENT.value}"
                             f" not found in dataset.")

    if BilevelDataCol.BUDGET not in dataset:
        raise AttributeError(f"{BilevelDataCol.BUDGET.value}"
                             f" not found in dataset.")

    nodes = list(dataset[DataCol.NODE_ID].values)

    weights = dataset[BilevelDataCol.TRANSPORT_WEIGHT_CLIENT]
    budgets = dataset[BilevelDataCol.BUDGET]

    lagrange_map = {}
    for i in nodes:
        for j in nodes:
            if i != j:
                dict_a = {z: a for (i,j,z), a
                          in weights(i,j).items()}
                dict_b = {z: budget for (i,j,z), budget
                          in budgets(i,j).items()}

                lagrange = self.calculate_lagrange(dict_a, dict_b)
                lagrange_record = {(i,j,z): lagrange[z] for z in lagrange.keys()}
                lagrange_map.update(lagrange_record)


    dataset.add_entity(name=BilevelDataCol.LAGRANGE,
                       keys=[DataCol.START_NODE,
                             DataCol.END_NODE,
                             BilevelDataCol.CLIENT_ID_ROUTE],
                       data_map=lagrange_map)

`LinearClientRanker` ¶

Bases: EntityProcessor

Sort and index clients by budget-to-weight ratio on each route.

For each route $(i,j)$, clients are sorted in descending order of $b_{ij}^z / a_{ij}^z$ and assigned zero-based indices $z = 0, 1, 2, \dots$. This ranking is essential for the [PrecendenceConstraint][oracle_paper.constraints.precedence_constraint.PrecendenceConstraint] and all Lagrange multiplier calculations.

Replaces the raw client entities (keyed by original client ID) with re-indexed entities keyed by $(i,j,z)$ tuples.

`process(dataset)` ¶

Rank clients and re-index entities by $(i,j,z)$.

Parameters:

Name	Type	Description	Default
`dataset`	`MultiEntityDataset`	Dataset with client routes, budgets, and weights (modified in-place).	required

Source code in src/oracle_paper/data/processor/client_ranker.py

def process(self, dataset: MultiEntityDataset) -> None:
    """Rank clients and re-index entities by $(i,j,z)$.

    Args:
        dataset: Dataset with client routes, budgets, and weights
            (modified in-place).
    """
    client_routes = dataset[BilevelDataCol.CLIENT_ROUTE]
    budgets = dataset[BilevelDataCol.BUDGET]
    weights = dataset[BilevelDataCol.TRANSPORT_WEIGHT_CLIENT]


    route_buckets = defaultdict(list)
    for (c_id,), (i, j) in client_routes.items():
        route_buckets[(i, j)].append(c_id)

    # mapping from client key -> (i,j,z)
    math_keys: Dict[Tuple, tuple] = {}

    math_weights: Dict[Tuple, float] = {}
    math_budgets: Dict[Tuple, float] = {}
    math_ratios: Dict[Tuple, float] = {}
    math_client_ids: Dict[Tuple, int] = {}

    for (i, j), clients in route_buckets.items():
        client_ratios = []
        for c_id in clients:
            ratio = budgets[c_id] / weights[c_id]
            client_ratios.append((c_id, ratio))

        client_ratios.sort(key=lambda x: x[1], reverse=True)


        for z, (c_id, ratio) in enumerate(client_ratios):
            math_key = (i, j, z)

            math_keys[(c_id,)] = math_key

            math_weights[math_key] = weights[c_id]
            math_budgets[math_key] = budgets[c_id]
            math_ratios[math_key] = ratio
            math_client_ids[math_key] = z


    math_indices = [DataCol.START_NODE,
                             DataCol.END_NODE,
                             BilevelDataCol.CLIENT_ID_ROUTE]

    dataset.add_entity(BilevelDataCol.CLIENT_KEY, [BilevelDataCol.CLIENT_KEY], math_keys)

    dataset.add_entity(BilevelDataCol.TRANSPORT_WEIGHT_CLIENT, math_indices, math_weights)
    dataset.add_entity(BilevelDataCol.BUDGET, math_indices, math_budgets)
    dataset.add_entity(BilevelDataCol.CLIENT_RATIO, math_indices, math_ratios)

    dataset.add_entity(BilevelDataCol.CLIENT_ID_ROUTE, math_indices, math_client_ids)

`RecursiveLagrangeCalculator()` ¶

Bases: TrackableProcessor

Compute recursive Lagrange multipliers for the PC-HLP model.

Groups clients on the same route $(i,j)$ that have already been sorted by LinearClientRanker, then merges adjacent client segments where the Lagrange-to-weight ratio is non-increasing. The merged groups form aggregated clients indexed by $(i,j,z)$ where $z$ is now a group index.

Adds four entities to the dataset:

Source code in src/oracle_paper/data/calculator/base.py

def __init__(self):
	"""Initialize the processor with metadata storage."""
	super().__init__()
	self._processor_metadata: dict[str, Any] = {}

`process(dataset)` ¶

Compute recursive Lagrange multipliers.

Parameters:

Name	Type	Description	Default
`dataset`	`MultiEntityDataset`	Dataset with Lagrange multipliers, weights, budgets, and client keys (modified in-place).	required

Raises:

Type	Description
`AttributeError`	If required entities are missing.

Source code in src/oracle_paper/data/calculator/recursive_lagrange.py

@track_metric("recursive_lagrange_time")
def process(self, dataset: MultiEntityDataset) -> None:
    """Compute recursive Lagrange multipliers.

    Args:
        dataset: Dataset with Lagrange multipliers, weights, budgets,
            and client keys (modified in-place).

    Raises:
        AttributeError: If required entities are missing.
    """
    if BilevelDataCol.TRANSPORT_WEIGHT_CLIENT not in dataset:
        raise AttributeError(f"{BilevelDataCol.TRANSPORT_WEIGHT_CLIENT.value}"
                             f" not found in dataset.")

    if BilevelDataCol.LAGRANGE not in dataset:
        raise AttributeError(f"{BilevelDataCol.LAGRANGE.value}"
                             f" not found in dataset.")

    nodes = list(dataset[DataCol.NODE_ID].values)

    lagrange = dataset[BilevelDataCol.LAGRANGE]
    weights = dataset[BilevelDataCol.TRANSPORT_WEIGHT_CLIENT]
    budget = dataset[BilevelDataCol.BUDGET]
    client_keys = dataset[BilevelDataCol.CLIENT_KEY]


    lagrange_map = {}
    keys_map = {}

    summed_weights_map = {}
    summed_budgets_map = {}
    for i in nodes:
        for j in nodes:
            if i != j:
                dict_a = {z: a for (_,__,z), a
                          in weights(i,j).items()}
                dict_lagrange = {z: l for (_,__,z), l
                          in lagrange(i,j).items()}
                dict_keys = {z: key for (key,), (cur_i,cur_j,z) in client_keys.items() if i== cur_i and j== cur_j}

                new_lagrange, new_indices = self.sort_lagrange_multipliers_dict(dict_lagrange, dict_keys, dict_a)


                lagrange_record = {(i,j,z): new_lagrange[z] for z in new_lagrange.keys()}
                keys_record = {(i,j,z) : new_indices[z] for z in new_indices.keys()}


                lagrange_map.update(lagrange_record)
                keys_map.update(keys_record)

    for (i,j,z), keys  in keys_map.items():
        summed_a = 0.0
        summed_b = 0.0
        for key in keys:
            original_i, original_j, original_z = client_keys[key]
            summed_a += weights[original_i, original_j, original_z]
            summed_b += budget[original_i, original_j, original_z]


        summed_weights_map[(i,j,z)] = summed_a
        summed_budgets_map[(i,j,z)] = summed_b


    dataset.add_entity(name=BilevelDataCol.RECURSIVE_LAGRANGE,
                       keys=[DataCol.START_NODE,
                             DataCol.END_NODE,
                             BilevelDataCol.CLIENT_ID_ROUTE],
                       data_map=lagrange_map
                       )

    dataset.add_entity(name=BilevelDataCol.CLIENT_KEYS,
                       keys=[DataCol.START_NODE,
                             DataCol.END_NODE,
                             BilevelDataCol.CLIENT_ID_ROUTE],
                       data_map=keys_map)

    dataset.add_entity(name=BilevelDataCol.SUMMED_LINEAR_WEIGHTS,
                       keys=[DataCol.START_NODE,
                             DataCol.END_NODE,
                             BilevelDataCol.CLIENT_ID_ROUTE],
                       data_map=summed_weights_map)

    dataset.add_entity(name=BilevelDataCol.SUMMED_BUDGETS,
                       keys=[DataCol.START_NODE,
                             DataCol.END_NODE,
                             BilevelDataCol.CLIENT_ID_ROUTE],
                       data_map=summed_budgets_map)

`sort_lagrange_multipliers_dict(dict_lagrange, dict_keys, dict_a)` `staticmethod` ¶

Merge adjacent clients where Lagrange/weight ratio is non-increasing.

Uses a stack-based algorithm: iterates over sorted clients and merges when $\lambda_k/a_k > \lambda_{k+1}/a_{k+1}$.

Parameters:

Name	Type	Description	Default
`dict_lagrange`	`dict[int, float]`	$\{z: \lambda_{ij}^z\}$.	required
`dict_keys`	`dict[int, int]`	$\{z: \text{original client IDs}\}$.	required
`dict_a`	`dict[int, float]`	$\{z: a_{ij}^z\}$.	required

Returns:

Type	Description
`dict[int, float]`	`(new_lagrange, new_ids)` — merged Lagrange multipliers
`dict[int, list[int]]`	and grouped client ID lists.

Source code in src/oracle_paper/data/calculator/recursive_lagrange.py

@staticmethod
def sort_lagrange_multipliers_dict(
    dict_lagrange: dict[int, float],
    dict_keys: dict[int, int],
    dict_a: dict[int, float],
) -> tuple[dict[int, float], dict[int, list[int]]]:
    r"""Merge adjacent clients where Lagrange/weight ratio is non-increasing.

    Uses a stack-based algorithm: iterates over sorted clients and
    merges when $\lambda_k/a_k > \lambda_{k+1}/a_{k+1}$.

    Args:
        dict_lagrange: $\{z: \lambda_{ij}^z\}$.
        dict_keys: $\{z: \text{original client IDs}\}$.
        dict_a: $\{z: a_{ij}^z\}$.

    Returns:
        ``(new_lagrange, new_ids)`` — merged Lagrange multipliers
        and grouped client ID lists.
    """


    stack =[]
    for k in range(len(dict_lagrange)):
        current_lagrange = dict_lagrange[k]
        current_a = dict_a[k]
        current_key =  [dict_keys[k]]

        current_ratio = current_lagrange / current_a

        while stack:
            prev_lagrange, prev_a, prev_key, prev_ratio = stack[-1]

            if prev_ratio >= current_ratio:
                break

            # else: prev_ration < current_ration
            # in this case we need to merge the lagrange multipliers
            # pop the first element in the stack
            # and replace with later with stack.append((current_l, current_a, current_key, current_ration))
            current_lagrange += prev_lagrange
            current_a += prev_a
            prev_key.extend(current_key)
            current_key = prev_key

            current_ratio = current_lagrange / current_a

            stack.pop()

        stack.append((current_lagrange, current_a, current_key, current_ratio))

    if not stack:
        return {}, {}

    res_lagrange, res_a, res_key, res_ratio = zip(*stack)

    new_lagrange = {new_z: res_lagrange[new_z] for new_z
                    in range(len(res_lagrange))}

    new_ids = {new_z: res_key[new_z] for new_z
               in range(len(res_key))}

    return new_lagrange, new_ids

Name	Type	Description	Default
`dict_a`	`dict[int, float]`	Client weights \(\{z: a_{ij}^z\}\) sorted by \(z\).	required
`dict_b`	`dict[int, float]`	Client budgets \(\{z: b_{ij}^z\}\) sorted by \(z\).	required

Name	Type	Description	Default
`dict_lagrange`	`dict[int, float]`	\(\{z: \lambda_{ij}^z\}\).	required
`dict_keys`	`dict[int, int]`	\(\{z: \text{original client IDs}\}\).	required
`dict_a`	`dict[int, float]`	\(\{z: a_{ij}^z\}\).	required

Data

BilevelClientGenerator(clients_per_route, random_count=False, min_budget_factor=1.2, max_budget_factor=1.8, possible_weights=None, seed=42) ¶

LagrangeCalculator() ¶

calculate_lagrange(dict_a, dict_b) staticmethod ¶

process(dataset) ¶

LinearClientRanker ¶

process(dataset) ¶

RecursiveLagrangeCalculator() ¶

process(dataset) ¶

sort_lagrange_multipliers_dict(dict_lagrange, dict_keys, dict_a) staticmethod ¶

`BilevelClientGenerator(clients_per_route, random_count=False, min_budget_factor=1.2, max_budget_factor=1.8, possible_weights=None, seed=42)` ¶

`LagrangeCalculator()` ¶

`calculate_lagrange(dict_a, dict_b)` `staticmethod` ¶

`process(dataset)` ¶

`LinearClientRanker` ¶

`process(dataset)` ¶

`RecursiveLagrangeCalculator()` ¶

`process(dataset)` ¶

`sort_lagrange_multipliers_dict(dict_lagrange, dict_keys, dict_a)` `staticmethod` ¶