Skip to content

Lagrange

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 
18
19
20
21
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 
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
@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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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
@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)