Calculator

Lagrange multiplier calculators and the trackable-processor base class.

`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)

`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

`TrackableProcessor()` ¶

Bases: EntityProcessor

Mixin for entity processors that track calculation metrics.

Provides set_metric / get_metric / get_all_metrics so processors can record timing and other data that flows into benchmark results and solution metadata.

Initialize the processor with metadata storage.

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] = {}

`get_all_metrics()` ¶

Get all recorded metrics.

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

def get_all_metrics(self) -> dict[str, Any]:
	"""Get all recorded metrics."""
	return self._processor_metadata.copy()

`get_metric(key)` ¶

Retrieve a metric value.

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

def get_metric(self, key: str) -> Any:
	"""Retrieve a metric value."""
	return self._processor_metadata.get(key)

`set_metric(key, value)` ¶

Store a metric value.

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

def set_metric(self, key: str, value: Any) -> None:
	"""Store a metric value."""
	self._processor_metadata[key] = value

`track_metric(metric_name)` ¶

Decorator to automatically track calculation time for a method.

Parameters:

Name	Type	Description	Default
`metric_name`	`str`	The name to store the metric under (e.g., "lagrange_time")	required

Example

@track_metric("lagrange_calculation_time") def process(self, dataset): # ... calculation code ...

The decorated method must be part of a TrackableProcessor instance. The elapsed time will be stored via set_metric(metric_name, elapsed_time).

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

def track_metric(metric_name: str) -> Callable:
	"""
	Decorator to automatically track calculation time for a method.

	Args:
		metric_name: The name to store the metric under (e.g., "lagrange_time")

	Example:
		@track_metric("lagrange_calculation_time")
		def process(self, dataset):
			# ... calculation code ...

	The decorated method must be part of a TrackableProcessor instance.
	The elapsed time will be stored via set_metric(metric_name, elapsed_time).
	"""

	def decorator(func: Callable) -> Callable:
		@wraps(func)
		def wrapper(self, *args, **kwargs) -> Any:
			start_time = time.time()
			try:
				result = func(self, *args, **kwargs)
				return result
			finally:
				elapsed_time = time.time() - start_time
				# Only store if the instance supports metrics (TrackableProcessor)
				if hasattr(self, "set_metric"):
					self.set_metric(metric_name, elapsed_time)

		return wrapper

	return decorator

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

Calculator

LagrangeCalculator() ¶

calculate_lagrange(dict_a, dict_b) staticmethod ¶

process(dataset) ¶

RecursiveLagrangeCalculator() ¶

process(dataset) ¶

sort_lagrange_multipliers_dict(dict_lagrange, dict_keys, dict_a) staticmethod ¶

TrackableProcessor() ¶

get_all_metrics() ¶

get_metric(key) ¶

set_metric(key, value) ¶

track_metric(metric_name) ¶

`LagrangeCalculator()` ¶

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

`process(dataset)` ¶

`RecursiveLagrangeCalculator()` ¶

`process(dataset)` ¶

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

`TrackableProcessor()` ¶

`get_all_metrics()` ¶

`get_metric(key)` ¶

`set_metric(key, value)` ¶

`track_metric(metric_name)` ¶