#!/usr/bin/env python3 """Correctness tests for the paper's Section 5 / Appendix C application models. Every transcription and every closed form used by run_apub_applications.py is pinned here against an independent computation, so a silent transcription bug cannot pass as a result. Run: python3 code/test_apub_paper_models.py """ import json import sys import numpy as np from scipy import stats sys.path.insert(0, __file__.rsplit("/", 1)[0]) import apub_paper_models as P rng = np.random.default_rng(20260802) RESULTS = [] def check(name, ok, detail): RESULTS.append({"test": name, "pass": bool(ok), "detail": detail}) print(f"[{'PASS' if ok else 'FAIL'}] {name}: {detail}") return ok # --------------------------------------------------------------------------- # T1 positive-stable draw has Laplace transform exp(-t^a) # --------------------------------------------------------------------------- def t1(): worst = 0.0 for lam in (2.0, 5.0): a = 1.0 / lam s = P.positive_stable(a, size=400000, rng=rng) for t in (0.3, 1.0, 3.0): emp = np.mean(np.exp(-t * s)) worst = max(worst, abs(emp - np.exp(-t ** a))) return check("positive_stable_laplace_transform", worst < 5e-3, f"max |E[exp(-tS)] - exp(-t^a)| = {worst:.2e} over lam in " f"{{2,5}}, t in {{0.3,1,3}} (400k draws)") # --------------------------------------------------------------------------- # T2 Gumbel copula: uniform marginals and Kendall tau = 1 - 1/lambda # --------------------------------------------------------------------------- def t2(): ok, det = True, [] for lam in (2.0, 5.0): u = P.gumbel_copula_uniforms(60000, 4, lam, rng) ks = max(stats.kstest(u[:, k], "uniform").statistic for k in range(4)) tau = stats.kendalltau(u[:2000, 0], u[:2000, 1]).statistic target = 1.0 - 1.0 / lam ok &= (ks < 0.01) and (abs(tau - target) < 0.03) det.append(f"lam={lam}: max KS={ks:.4f}, tau={tau:.3f} (theory {target:.3f})") return check("gumbel_copula_marginals_and_kendall_tau", ok, "; ".join(det)) # --------------------------------------------------------------------------- # T3 closed-form product-mix recourse == recourse LP # --------------------------------------------------------------------------- def t3(): xi = P.sample_product_mix_xi(40, rng) worst = 0.0 for _ in range(6): x = rng.uniform(0, 12, size=20) cf, feas = P.recourse_pm_closed_form(x, xi) assert feas.all(), "closed form only claims optimality where sum_j y_j <= h2" for n in range(0, 40, 7): lp = P.recourse_pm_lp(x, xi["q"][n], xi["w"][n], xi["h1"][n], xi["h2"][n]) worst = max(worst, abs(cf[n] - lp) / max(1.0, abs(lp))) return check("recourse_closed_form_equals_LP", worst < 1e-8, f"max relative gap over 6 x-vectors x 6 scenarios = {worst:.2e}") # --------------------------------------------------------------------------- # T4 APUB(alpha) is the CVaR of the bootstrap means and its t-formulation # agrees with the sorted-tail formula # --------------------------------------------------------------------------- def t4(): theta = rng.gamma(2.0, 3.0, size=40) V = P.bootstrap_multiplicities(40, 4000, rng) zm = (V @ theta) / 40 worst = 0.0 for alpha in (0.05, 0.2, 0.5, 1.0): direct = P.apub_from_costs(theta, V, alpha) grid = np.linspace(zm.min() - 1, zm.max() + 1, 200001) var = np.quantile(zm, 1 - alpha) if alpha < 1 else zm.min() - 1 cand = np.unique(np.concatenate([grid[::200], [var]])) vals = cand + (np.maximum(zm[None, :] - cand[:, None], 0).mean(axis=1)) / alpha worst = max(worst, abs(direct - vals.min()) / max(1.0, abs(direct))) return check("apub_equals_inf_t_formulation", worst < 1e-3, f"max relative gap between sorted-tail CVaR and " f"inf_t {{t + E[(Z-t)_+]/alpha}} = {worst:.2e}") # --------------------------------------------------------------------------- # T5 APUB monotone decreasing in alpha, and alpha = 1 reduces to SAA # (Remark rem:APUB-DEF-OPT), on the real 20x8 model # --------------------------------------------------------------------------- def t5(): xi = P.sample_product_mix_xi(25, rng) V = P.bootstrap_multiplicities(25, 400, rng) objs = {} for alpha in (0.05, 0.2, 0.5, 1.0): objs[alpha] = P.solve_pm_random_recourse(xi, alpha, V, x_ub=5000.0)["obj"] saa = P.solve_pm_random_recourse(xi, 1.0, None, x_ub=5000.0)["obj"] mono = all(objs[a] >= objs[b] - 1e-6 for a, b in zip([0.05, 0.2, 0.5], [0.2, 0.5, 1.0])) # Remark rem:APUB-DEF-OPT holds for the EXACT bootstrap law; with a finite # M the alpha = 1 model is the Monte-Carlo bootstrap approximation of SAA, # so the right check is that the gap contracts as M grows. gaps = [] for M in (200, 1600, 12800): Vm = P.bootstrap_multiplicities(25, M, rng) o = P.solve_pm_random_recourse(xi, 1.0, Vm, x_ub=5000.0)["obj"] gaps.append(abs(o - saa) / abs(saa)) # At alpha = 1 the model reduces to a bootstrap-REWEIGHTED SAA with weights # w_n = (1/M) sum_m V_mn (mean 1, sd ~ M^{-1/2}), so the gap to plain SAA is # O(M^{-1/2}), not zero at finite M. contracts = all(gaps[i] > gaps[i + 1] for i in range(len(gaps) - 1)) \ and gaps[-1] < 1e-2 return check("apub_sp_monotone_in_alpha_and_alpha1_converges_to_SAA", mono and contracts, f"objs by alpha = { {k: round(v, 4) for k, v in objs.items()} } " f"(monotone non-increasing in alpha: {mono}); SAA = {saa:.4f}; " f"alpha=1 relative gap to SAA at M=(200,1600,12800) = " f"{[f'{g:.2e}' for g in gaps]}") # --------------------------------------------------------------------------- # T6 the APUB-SP LP optimum equals the APUB of the recourse costs evaluated # independently at the returned x (i.e. the LP really optimises the paper's # objective, not a relaxation of it) # --------------------------------------------------------------------------- def t6(): xi = P.sample_product_mix_xi(25, rng) V = P.bootstrap_multiplicities(25, 400, rng) det = [] ok = True for alpha in (0.1, 0.5): sol = P.solve_pm_random_recourse(xi, alpha, V, x_ub=5000.0) cf, _ = P.recourse_pm_closed_form(sol["x"], xi) rebuilt = float(P.PM_C @ sol["x"]) + P.apub_from_costs(cf, V, alpha) gap = abs(rebuilt - sol["obj"]) / max(1.0, abs(sol["obj"])) ok &= gap < 1e-6 det.append(f"alpha={alpha}: LP obj={sol['obj']:.4f}, " f"independent rebuild={rebuilt:.4f}, rel gap={gap:.2e}") return check("apub_sp_LP_objective_matches_independent_rebuild", ok, "; ".join(det)) # --------------------------------------------------------------------------- # T7 fixed-recourse closed form == the W y = h - T x recourse LP # --------------------------------------------------------------------------- def t7(): from scipy.optimize import linprog gam = P.sample_fixed_recourse_gamma(30, rng) worst = 0.0 for _ in range(8): x = rng.uniform(0, 200, size=4) cf = P.fr_recourse(x, gam) Tg = P.FR_T_BASE - gam[:, :, None] * 0.25 # (n,2,4) for n in range(0, 30, 5): rhs = 500.0 * gam[n] - Tg[n] @ x res = linprog(P.FR_QCOST, A_eq=P.FR_W, b_eq=rhs, bounds=[(0, None)] * 4, method="highs") worst = max(worst, abs(cf[n] - res.fun) / max(1.0, abs(res.fun))) return check("fixed_recourse_closed_form_equals_LP", worst < 1e-8, f"max relative gap over 8 x-vectors x 6 scenarios = {worst:.2e}") # --------------------------------------------------------------------------- # T8 the DRO Lipschitz moduli are the true ones (finite-difference check) # --------------------------------------------------------------------------- def t8(): ok, det = True, [] for _ in range(5): x = rng.uniform(0, 200, size=4) g = rng.uniform(0.5, 20.0, size=(4000, 2)) f = P.fr_recourse(x, g) pert = g + rng.uniform(-1e-3, 1e-3, size=g.shape) num = np.abs(P.fr_recourse(x, pert) - f) / np.abs(pert - g).sum(axis=1) ok &= num.max() <= P.fr_lipschitz(x) * (1 + 1e-6) det.append(f"emp={num.max():.2f} <= analytic={P.fr_lipschitz(x):.2f}") # newsvendor: modulus is max(h, b) and is x-independent for _ in range(3): x = rng.uniform(20, 90, size=10) d = rng.uniform(20, 90, size=(4000, 10)) pert = d + rng.uniform(-1e-3, 1e-3, size=d.shape) num = np.abs(P.nv_cost(x, pert) - P.nv_cost(x, d)) / \ np.abs(pert - d).sum(axis=1) ok &= num.max() <= max(P.NV_H, P.NV_B) * (1 + 1e-6) det.append(f"newsvendor modulus max(h,b)={max(P.NV_H, P.NV_B)} respected") return check("wasserstein_lipschitz_moduli_are_exact", ok, "; ".join(det)) # --------------------------------------------------------------------------- # T9 newsvendor SAA LP optimum equals the direct empirical-average minimiser # (checked against the closed-form critical quantile, product by product) # --------------------------------------------------------------------------- def t9(): xi = P.sample_newsvendor(400, rng, case=1) sol = P.solve_nv(xi, alpha=1.0, V=None) # The multi-product newsvendor separates by product; the marginal of # p x + h(x-d)_+ + b(d-x)_+ is p + hF(x) - b(1-F(x)), so the empirical # minimiser is the (b - p)/(h + b) empirical quantile of that column. qlev = (P.NV_B - P.NV_P) / (P.NV_H + P.NV_B) closed = np.quantile(xi, qlev, axis=0, method="inverted_cdf") obj_closed = float(P.nv_cost(closed, xi).mean()) gap = abs(obj_closed - sol["obj"]) / max(1.0, abs(sol["obj"])) return check("newsvendor_SAA_LP_equals_empirical_quantile_solution", gap < 1e-9, f"LP obj={sol['obj']:.6f}, closed-form quantile obj=" f"{obj_closed:.6f}, rel gap={gap:.2e}, " f"max |x_LP - x_quantile| = {np.abs(sol['x'] - closed).max():.3f}") # --------------------------------------------------------------------------- # T10 newsvendor APUB-SP LP objective matches an independent rebuild, and the # DRO shift is exactly eps*max(h,b) with an unchanged solution # --------------------------------------------------------------------------- def t10(): xi = P.sample_newsvendor(30, rng, case=1) V = P.bootstrap_multiplicities(30, 800, rng) sol = P.solve_nv(xi, alpha=0.1, V=V) theta = P.nv_cost(sol["x"], xi) rebuilt = P.apub_from_costs(theta, V, 0.1) gap = abs(rebuilt - sol["obj"]) / max(1.0, abs(sol["obj"])) saa = P.solve_nv(xi, alpha=1.0, V=None) dro = P.solve_nv(xi, alpha=1.0, V=None, dro_eps=0.7) shift_ok = abs((dro["obj"] - saa["obj"]) - 0.7 * max(P.NV_H, P.NV_B)) < 1e-6 same_x = np.abs(dro["x"] - saa["x"]).max() < 1e-6 return check("newsvendor_apub_rebuild_and_dro_constant_shift", gap < 1e-6 and shift_ok and same_x, f"APUB LP obj={sol['obj']:.6f} vs rebuild={rebuilt:.6f} " f"(rel gap {gap:.2e}); DRO-SAA shift=" f"{dro['obj'] - saa['obj']:.6f} vs eps*max(h,b)=" f"{0.7 * max(P.NV_H, P.NV_B):.6f}; x unchanged={same_x}") # --------------------------------------------------------------------------- # T11 fixed-recourse DRO solution DOES move with eps (decision-dependent # Lipschitz modulus), unlike the newsvendor # --------------------------------------------------------------------------- def t11(): gam = P.sample_fixed_recourse_gamma(120, rng) saa = P.solve_fr(gam, 1.0, None) moves = [] eps_grid = (0.05, 0.2, 1.0) for eps in eps_grid: dro = P.solve_fr(gam, 1.0, None, dro_eps=eps) moves.append(float(np.linalg.norm(dro["x"] - saa["x"]))) nondec = all(moves[i] <= moves[i + 1] + 1e-6 for i in range(len(moves) - 1)) ok = moves[-1] > 1e-6 and nondec return check("fixed_recourse_DRO_solution_is_decision_dependent", ok, f"||x_DRO - x_SAA||_2 at eps={eps_grid} = " f"{[round(m, 3) for m in moves]} (non-decreasing and eventually " f"positive => the Wasserstein modulus really depends on x; " f"contrast the newsvendor, where it cannot move at all)") def main(): fns = [t1, t2, t3, t4, t5, t6, t7, t8, t9, t10, t11] allok = True for f in fns: allok &= bool(f()) out = {"all_passed": bool(allok), "tests": RESULTS} with open(__file__.rsplit("/", 2)[0] + "/results/paper_model_unit_tests.json", "w") as fh: json.dump(out, fh, indent=1) print("\nALL PASSED" if allok else "\nSOME FAILED") return 0 if allok else 1 if __name__ == "__main__": sys.exit(main())