Spaces:
Running
Running
File size: 4,992 Bytes
b2e68d2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 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 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 | #!/usr/bin/env python3
"""Exact finite audit of Theorem 4 (the matching EMD lower bound).
The source construction has two truths. They differ only by moving mass
epsilon from day 4 to day 2, while a midpoint prediction is equally close to
both. A deterministic ski-rental policy is a threshold K (or infinity).
Enumerating every threshold proves the two incompatible threshold regions;
the final mixture inequality is Yao's argument for randomized policies.
"""
from __future__ import annotations
import argparse
import json
from fractions import Fraction
from pathlib import Path
INF = None
def distribution_pair(b: int) -> tuple[dict[int, Fraction], dict[int, Fraction], dict[int, Fraction], Fraction]:
delta = Fraction(1, 8 * b)
epsilon = Fraction(1, 2 * b) + delta
p1 = {
1: Fraction(1, 2),
2: Fraction(1, 4),
4: epsilon,
b + 3: Fraction(1, 4) - epsilon,
}
p2 = {
1: Fraction(1, 2),
2: Fraction(1, 4) + epsilon,
b + 3: Fraction(1, 4) - epsilon,
}
midpoint = {
1: Fraction(1, 2),
2: Fraction(1, 4) + epsilon / 2,
4: epsilon / 2,
b + 3: Fraction(1, 4) - epsilon,
}
return p1, p2, midpoint, epsilon
def policy_cost(distribution: dict[int, Fraction], threshold: int | None, b: int) -> Fraction:
if threshold is INF:
return sum(Fraction(day) * mass for day, mass in distribution.items())
return sum(
(Fraction(day) if day <= threshold else Fraction(threshold + b)) * mass
for day, mass in distribution.items()
)
def emd(first: dict[int, Fraction], second: dict[int, Fraction]) -> Fraction:
cumulative = Fraction(0)
distance = Fraction(0)
points = sorted(set(first) | set(second))
for left, right in zip(points, points[1:]):
cumulative += first.get(left, 0) - second.get(left, 0)
distance += abs(cumulative) * (right - left)
return distance
def audit_one(b: int) -> dict:
p1, p2, midpoint, epsilon = distribution_pair(b)
candidates: list[int | None] = list(range(0, b + 4)) + [INF]
costs1 = {k: policy_cost(p1, k, b) for k in candidates}
costs2 = {k: policy_cost(p2, k, b) for k in candidates}
opt1 = min(costs1.values())
opt2 = min(costs2.values())
opt1_thresholds = [k for k, value in costs1.items() if value == opt1]
opt2_thresholds = [k for k, value in costs2.items() if value == opt2]
# Buying after the last support point ties rent-forever; the source's
# canonical choice is A_infinity and it is present in the optimum set.
assert INF in opt1_thresholds, opt1_thresholds
assert opt2_thresholds == [2], opt2_thresholds
low = [costs1[k] - opt1 for k in range(4)]
high = [costs2[k] - opt2 for k in candidates if k is INF or k >= 4]
min_low = min(low)
min_high = min(high)
randomized_worst_case = min(min_low, min_high) / 2
midpoint_emd_1 = emd(midpoint, p1)
midpoint_emd_2 = emd(midpoint, p2)
assert midpoint_emd_1 == epsilon == midpoint_emd_2
# For q=P(threshold <= 3), p1 loss >= q*min_low and p2 loss >=
# (1-q)*min_high. Their maximum is at least half the smaller constant.
assert randomized_worst_case > 0
return {
"b": b,
"delta": str(Fraction(1, 8 * b)),
"epsilon": str(epsilon),
"support_max": b + 3,
"prediction_emd_to_p1": str(midpoint_emd_1),
"prediction_emd_to_p2": str(midpoint_emd_2),
"p1_optimal_threshold": "infinity",
"p2_optimal_threshold": 2,
"low_thresholds_0_to_3_loss": [str(value) for value in low],
"min_loss_p1_for_K_le_3": str(min_low),
"min_loss_p2_for_K_ge_4_or_infinity": str(min_high),
"randomized_worst_case_lower_bound": str(randomized_worst_case),
"randomized_bound_over_b_times_emd": float(randomized_worst_case / (b * epsilon)),
"enumerated_deterministic_policies": len(candidates),
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
# These are finite support instances, with a nontrivial range of b.
rows = [audit_one(b) for b in (16, 32, 64, 128, 256, 1024, 4096, 16384, 65536, 262144)]
result = {
"theorem": "Theorem 4",
"arithmetic": "exact Fraction",
"rows": rows,
"all_midpoint_emd_equal_epsilon": all(
r["prediction_emd_to_p1"] == r["prediction_emd_to_p2"] == r["epsilon"] for r in rows
),
"all_optima_match_source": all(
r["p1_optimal_threshold"] == "infinity" and r["p2_optimal_threshold"] == 2 for r in rows
),
"all_randomized_lower_bounds_positive": all(
Fraction(r["randomized_worst_case_lower_bound"]) > 0 for r in rows
),
}
args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
print(json.dumps(result, indent=2))
if __name__ == "__main__":
main()
|