code/test_apub_paper_models.py

#!/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())