Skip to content

Compare

Four helpers for questions that sit next to a decision rather than inside it: did the distribution of outcomes move between two periods (compare_outcome_laws), could an A/B split have happened by chance (sample_two_row), which single noisy feature best predicts a target (best_coordinate), and when to accept an item from a stream you cannot go back to (single_item_max_of_samples, single_item_one_sample). None of them is called by Module.decide.

Public here: compare_outcome_laws, sample_two_row, best_coordinate, single_item_max_of_samples, single_item_one_sample (top level); w2_1d, sample_switch and apply_coordinate from the submodules. Recipes: Compare outcomes and features. Read the honest limits on each helper's theory page before quoting a factor (for example, factor_claimed=0.5 is a guarantee in expectation only for single_item_max_of_samples, under its stated assumptions; the scalar-bar rule reports None: prophet inequalities). sample_two_row and sample_switch take limits= and raise ValueError above max_table_cells_exact and max_switch_steps.

courtade_kumar

Courtade–Kumar 1-bit summaries (family 119, Lean).

After independent bit-flip noise on uniform bits, a coordinate (or its complement) retains the most mutual information. This module returns that coordinate instead of training a 1-bit bottleneck.

CoordinateSelection

Bases: TypedDict

Result of :func:best_coordinate.

best_coordinate 

best_coordinate(
    bits: ArrayLike, target: ArrayLike
) -> CoordinateSelection

bits shape (n_samples, d) in {0,1}. target length n_samples.

Source code in src/grounded_decision/compare/courtade_kumar.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def best_coordinate(bits: npt.ArrayLike, target: npt.ArrayLike) -> CoordinateSelection:
    """``bits`` shape (n_samples, d) in {0,1}. ``target`` length n_samples."""

    bit_matrix: npt.NDArray[np.int_] = np.asarray(bits, dtype=np.int_)
    labels: npt.NDArray[np.int_] = np.asarray(target, dtype=np.int_)
    if bit_matrix.ndim != 2 or len(labels) != bit_matrix.shape[0]:
        raise ValueError("bits must be (n, d) aligned with target")
    _n, d = bit_matrix.shape
    scores: list[float] = []
    for j in range(d):
        agree = _agreement(bit_matrix[:, j], labels)
        scores.append(max(agree, 1.0 - agree))
    j_star = int(np.argmax(scores))
    flipped = _agreement(bit_matrix[:, j_star], labels) < 0.5
    return {
        "index": j_star,
        "use_complement": flipped,
        "agreement": scores[j_star],
        "catalog": REF_COURTADE.to_dict(),
        "rule": "keep a coordinate; do not train a 1-bit mixer on noisy uniform bits",
    }

ot_stability

Wasserstein-2 comparison with the 1/3 Brenier stability (family 374).

Lean: ‖T_μ − T_ν‖_{L²} ≤ C W₂(μ,ν)^{1/3}, and every larger exponent fails. To keep map error ≤ τ, require W₂ ≲ τ³ (C is absorbed into a user scale).

OutcomeLawComparison

Bases: TypedDict

Result of :func:compare_outcome_laws.

w2_1d 

w2_1d(a: ArrayLike, b: ArrayLike) -> float

Exact W₂ between two empirical 1-D distributions of any sizes.

In one dimension the optimal transport plan matches quantiles. With sample sizes n and m, both empirical quantile functions are step functions; this integrates the squared difference exactly over the merged breakpoints k/n and j/m. Every sample counts: no tail is dropped when the sizes differ.

Source code in src/grounded_decision/compare/ot_stability.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def w2_1d(a: npt.ArrayLike, b: npt.ArrayLike) -> float:
    """Exact W₂ between two empirical 1-D distributions of any sizes.

    In one dimension the optimal transport plan matches quantiles. With
    sample sizes ``n`` and ``m``, both empirical quantile functions are
    step functions; this integrates the squared difference exactly over
    the merged breakpoints ``k/n`` and ``j/m``. Every sample counts: no
    tail is dropped when the sizes differ.
    """

    first = np.sort(np.asarray(a, dtype=float).ravel())
    second = np.sort(np.asarray(b, dtype=float).ravel())
    if len(first) == 0 or len(second) == 0:
        return 0.0
    breakpoints = np.union1d(
        np.arange(1, len(first) + 1) / len(first),
        np.arange(1, len(second) + 1) / len(second),
    )
    interval_widths = np.diff(np.concatenate(([0.0], breakpoints)))
    # The quantile function at level q is the ceil(q * size)-th smallest value.
    midpoints = breakpoints - interval_widths / 2
    first_quantiles = first[
        np.minimum(np.ceil(midpoints * len(first)).astype(int) - 1, len(first) - 1)
    ]
    second_quantiles = second[
        np.minimum(np.ceil(midpoints * len(second)).astype(int) - 1, len(second) - 1)
    ]
    return float(
        np.sqrt(np.sum(interval_widths * (first_quantiles - second_quantiles) ** 2))
    )

map_tolerance 

map_tolerance(tau: float, scale: float = 1.0) -> float

Largest W₂ compatible with Brenier-map error τ under exponent 1/3.

Source code in src/grounded_decision/compare/ot_stability.py 
49
50
51
52
def map_tolerance(tau: float, scale: float = 1.0) -> float:
    """Largest W₂ compatible with Brenier-map error τ under exponent 1/3."""

    return float(scale) * float(tau) ** 3

tables

Contingency-table sampling (family 115).

Exact sampling for 2-row tables via sequential hypergeometric draws. Those draws follow the hypergeometric (Fisher exact test) distribution on tables with the given margins, which is what a permutation test needs. It is not the uniform distribution on those tables. General tables: 2×2 switch chain, which is symmetric on the fiber and therefore uniform in stationarity. This is not the paper FPRAS; the mixer_status field says so.

TableSampleDict

Bases: TypedDict

Serialized form of :class:TableSample.

sample_two_row 

sample_two_row(
    col_sums: Sequence[int],
    row0: int,
    *,
    rng: Generator | None = None,
    limits: Limits = DEFAULT_LIMITS,
) -> TableSample

Draw one 2-row table with the given margins, exactly (hypergeometric).

Raises:

Type Description
ValueError

row0 is incompatible with the column sums, or the table has more than limits.max_table_cells_exact cells.
Source code in src/grounded_decision/compare/tables.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
100
101
102
def sample_two_row(
    col_sums: Sequence[int],
    row0: int,
    *,
    rng: np.random.Generator | None = None,
    limits: Limits = DEFAULT_LIMITS,
) -> TableSample:
    """Draw one 2-row table with the given margins, exactly (hypergeometric).

    Raises:
        ValueError: ``row0`` is incompatible with the column sums, or the
            table has more than ``limits.max_table_cells_exact`` cells.
    """

    cells = 2 * len(col_sums)
    if cells > limits.max_table_cells_exact:
        raise ValueError(
            f"{cells} cells exceed the exact-sampling limit "
            f"{limits.max_table_cells_exact}; use sample_switch"
        )
    rng = rng or np.random.default_rng()
    cols = [int(c) for c in col_sums]
    total = sum(cols)
    if not (0 <= row0 <= total):
        raise ValueError("row0 incompatible with column sums")
    top = _hypergeom(row0, cols, rng)
    bottom = [int(c) - a for c, a in zip(col_sums, top, strict=True)]
    if any(x < 0 for x in bottom):
        raise RuntimeError("hypergeometric produced a negative cell")
    return TableSample(
        table=(tuple(top), tuple(bottom)),
        method="two_row_hypergeometric",
        mixer_status="exact_hypergeometric",
    )

sample_switch 

sample_switch(
    row_sums: Sequence[int],
    col_sums: Sequence[int],
    *,
    steps: int = 2000,
    rng: Generator | None = None,
    limits: Limits = DEFAULT_LIMITS,
) -> TableSample

Run the 2×2 switch chain from a feasible table with the given margins.

Raises:

Type Description
ValueError

The margins disagree, or steps exceeds limits.max_switch_steps.
Source code in src/grounded_decision/compare/tables.py 
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
def sample_switch(
    row_sums: Sequence[int],
    col_sums: Sequence[int],
    *,
    steps: int = 2000,
    rng: np.random.Generator | None = None,
    limits: Limits = DEFAULT_LIMITS,
) -> TableSample:
    """Run the 2×2 switch chain from a feasible table with the given margins.

    Raises:
        ValueError: The margins disagree, or ``steps`` exceeds
            ``limits.max_switch_steps``.
    """

    if steps > limits.max_switch_steps:
        raise ValueError(
            f"{steps} switch steps exceed the limit {limits.max_switch_steps}"
        )
    rng = rng or np.random.default_rng()
    r = [int(x) for x in row_sums]
    c = [int(x) for x in col_sums]
    if sum(r) != sum(c):
        raise ValueError("row and column sums differ")
    # Greedy fill as a feasible start.
    table = np.zeros((len(r), len(c)), dtype=int)
    rr, cc = r[:], c[:]
    for i in range(len(r)):
        for j in range(len(c)):
            table[i, j] = min(rr[i], cc[j])
            rr[i] -= table[i, j]
            cc[j] -= table[i, j]
    m, n = table.shape
    if m < 2 or n < 2:
        return TableSample(
            tuple(tuple(int(x) for x in row) for row in table),
            "switch",
            "degenerate",
        )
    for _ in range(steps):
        i1, i2 = rng.choice(m, size=2, replace=False)
        j1, j2 = rng.choice(n, size=2, replace=False)
        a, b = table[i1, j1], table[i1, j2]
        c_, d = table[i2, j1], table[i2, j2]
        if rng.integers(2) == 0:
            if a > 0 and d > 0:
                table[i1, j1] -= 1
                table[i2, j2] -= 1
                table[i1, j2] += 1
                table[i2, j1] += 1
        else:
            if b > 0 and c_ > 0:
                table[i1, j2] -= 1
                table[i2, j1] -= 1
                table[i1, j1] += 1
                table[i2, j2] += 1
    return TableSample(
        tuple(tuple(int(x) for x in row) for row in table),
        method="switch_chain",
        mixer_status=f"switch_steps={steps}_not_paper_fpras",
    )

prophet

Single-item prophet rules: pick one arrival online, without hindsight.

Values arrive one at a time; each must be accepted or rejected on the spot, and at most one can be accepted. The "prophet" sees every value in advance and takes the largest. Two threshold rules are provided:

  • :func:single_item_max_of_samples - the bar is the largest of one past sample per arrival. This rule has a proven guarantee (Rubinstein, Wang and Weinberg, "Optimal single-choice prophet inequalities from samples", ITCS 2020): its expected value is at least half the prophet's expected value. factor_claimed is 0.5.
  • :func:single_item_one_sample - the bar is one scalar you choose (for example yesterday's median). For an arbitrary scalar bar there is no worst-case guarantee, so factor_claimed is None and the guarantee field says so. Measure the ratio on your own traffic.

The catalog's Lean-verified factor for the more general matroid setting (family 111) is 2^{-310}: true, and not a usable product factor. It is carried on every result as lean_factor so you can see it and leave it out of any SLO.

Why the max-of-samples bar earns 1/2

Assumptions: values X_1..X_n are independent and non-negative; S_i is one independent draw from the same distribution as X_i (for example yesterday's value at the same slot); the arrival order is fixed in advance; values have no ties (continuous distributions, or break ties with a tiny random perturbation). The rule accepts the first X_i >= T where T = max_i S_i.

Proof. Pool the 2n numbers and sort them, Z_1 > Z_2 > ... > Z_2n. Condition on the unordered pairs {X_i, S_i}. Because X_i and S_i are independent and identically distributed, which member of each pair is the value is a fair coin, independent across pairs. Let f be the position of the first sample in the sorted order. The values above T = Z_f are exactly Z_1..Z_{f-1}, so when f >= 2 the rule accepts one of them and earns at least Z_{f-1}; when f = 1 it earns at least 0. The prophet earns Z_v, v the position of the first value. Writing Z_k = sum_{j>=k} (Z_j - Z_{j+1}) (with Z_{2n+1} = 0), it suffices to show for every j: P(2 <= f <= j + 1) >= P(v <= j) / 2.

  • If positions 1..j hold j different pairs, P(v <= j) = 1 - 2^{-j} (some value among them), and P(2 <= f <= j + 1) = P(Z_1 is a value) - P(Z_1..Z_{j+1} are all values) >= 1/2 - 2^{-(j+1)}, which is exactly half.
  • Otherwise two of positions 1..j form a pair, so one is a value: P(v <= j) = 1; and Z_1..Z_{j+1} cannot all be values, so P(2 <= f <= j + 1) = P(Z_1 is a value) = 1/2.

Summing over j gives E[rule] >= E[Z_{f-1}; f >= 2] >= E[max X] / 2. If no value reaches the bar the rule takes the last arrival; with non-negative values that can only add. The factor cannot be improved: no online rule beats 1/2 in the worst case, even one that knows every distribution exactly.

LEAN_FACTOR module-attribute 

LEAN_FACTOR = 2 ** -310

Catalog family 111's Lean-verified factor; reported, never promised.

ProphetChoice

Bases: TypedDict

Result of :func:single_item_max_of_samples or :func:single_item_one_sample.

index and value are None only when values is empty. factor_claimed is the guaranteed fraction of the prophet's expected value, or None when the rule has no guarantee; guarantee says in words what is (or is not) promised and under which assumptions.

single_item_max_of_samples 

single_item_max_of_samples(
    values: Sequence[float], samples: Sequence[float]
) -> ProphetChoice

Take the first value that reaches the largest past sample.

Parameters:

Name Type Description Default
values Sequence[float]

Today's arrivals, in arrival order (non-negative).

 required
samples Sequence[float]

One past draw per arrival, from the same distribution as that arrival (for example yesterday's value at the same slot). Must have the same length as values and be non-empty.

 required

Returns:

Type Description
ProphetChoice

The choice, with factor_claimed = 0.5: in expectation the rule
ProphetChoice

earns at least half of what an all-knowing prophet would. The proof
ProphetChoice

and its assumptions are in this module's docstring.

Raises:

Type Description
ValueError

If samples is empty or its length differs from values.
Source code in src/grounded_decision/compare/prophet.py 
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
def single_item_max_of_samples(
    values: Sequence[float],
    samples: Sequence[float],
) -> ProphetChoice:
    """Take the first value that reaches the largest past sample.

    Args:
        values: Today's arrivals, in arrival order (non-negative).
        samples: One past draw per arrival, from the same distribution as
            that arrival (for example yesterday's value at the same slot).
            Must have the same length as ``values`` and be non-empty.

    Returns:
        The choice, with ``factor_claimed = 0.5``: in expectation the rule
        earns at least half of what an all-knowing prophet would. The proof
        and its assumptions are in this module's docstring.

    Raises:
        ValueError: If ``samples`` is empty or its length differs from
            ``values``.
    """

    if not samples:
        raise ValueError("single_item_max_of_samples needs one past sample per arrival")
    if len(samples) != len(values):
        raise ValueError(
            f"one past sample per arrival is required: {len(values)} values, "
            f"{len(samples)} samples"
        )
    bar = float(max(samples))
    index, chosen = _first_reaching(bar, values)
    return {
        "index": index,
        "value": chosen,
        "threshold_sample": bar,
        "factor_claimed": 0.5,
        "guarantee": MAX_OF_SAMPLES_GUARANTEE,
        "lean_factor": LEAN_FACTOR,
        "catalog": REF_PROPHET.to_dict(),
        "note": (
            "max-of-samples rule: the bar is the largest of one past sample per "
            "arrival; 1/2 is guaranteed in expectation under the stated "
            "assumptions; Lean 2^{-310} is not used as a product factor"
        ),
    }

single_item_one_sample 

single_item_one_sample(
    sample: float, values: Sequence[float]
) -> ProphetChoice

Irrevocable online choice: take the first value that reaches sample.

sample is one scalar bar. For an arbitrary bar there is no worst-case guarantee, so factor_claimed is None. When you have one past sample per arrival, use :func:single_item_max_of_samples, which guarantees half the prophet's expected value.

Source code in src/grounded_decision/compare/prophet.py 
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 single_item_one_sample(
    sample: float,
    values: Sequence[float],
) -> ProphetChoice:
    """Irrevocable online choice: take the first value that reaches ``sample``.

    ``sample`` is one scalar bar. For an arbitrary bar there is no
    worst-case guarantee, so ``factor_claimed`` is ``None``. When you have
    one past sample per arrival, use :func:`single_item_max_of_samples`,
    which guarantees half the prophet's expected value.
    """

    index, chosen = _first_reaching(sample, values)
    return {
        "index": index,
        "value": chosen,
        "threshold_sample": float(sample),
        "factor_claimed": None,
        "guarantee": SCALAR_BAR_GUARANTEE,
        "lean_factor": LEAN_FACTOR,
        "catalog": REF_PROPHET.to_dict(),
        "note": (
            "single-threshold rule with a scalar bar; no worst-case factor is "
            "claimed; Lean 2^{-310} is not used as a product factor"
        ),
    }