scripts/claim3_thm38_separation.py

"""
CLAIM 3 -- Theorem 3.8:  there exist hypothesis classes with gamma-graph
dimension d_gamma and gamma-OIG dimension at most 3 for which EVERY finite
aggregation algorithm (Def. 3.6) using an interpolating aggregation rule
(Def. 3.7) still needs n = Omega(d_gamma/eps), i.e. E[L] > eps for
n <= d_gamma/(128 eps) -- separating them from general learners that get
Otilde(1/eps).

Independent method
------------------
(a) Build the class explicitly: H = {h_A : A subset N, |A| = d}, h_A = 0 on A and
    a unique gamma_A in (gamma,1] elsewhere.  Compute d_gamma EXHAUSTIVELY and
    the gamma-OIG dimension EXHAUSTIVELY (all point sets of size k, all
    orientations of the induced one-inclusion hypergraph, plus vertex-induced
    subgraphs) -- both from Definitions 3.2 and 3.9.
(b) Simulate the hard distribution family and a *best-effort* finite aggregation
    algorithm (it puts every observed point in the zero-set of a selected
    hypothesis and spends the rest of its budget guessing), for a grid of
    (d, eps, m) and several interpolating rules.  Check E[L] > eps at
    n = floor(d/(128 eps)).
(c) Verify the separation numerically: run an explicit *general* (non-finite-
    aggregation) improper learner on the same class and measure its loss.

Seeds: numpy default_rng(20260725 + offset).  CPU only.
"""

import math
import os
import sys

import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from core import (  # noqa: E402
    dump_json,
    gamma_graph_dim,
    gamma_oig_dim,
    is_interpolating_rule,
    rule_max,
    rule_mean,
    rule_median,
    rule_midrange,
    rule_min,
    thm38_class,
)

OUT = os.path.join(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "outputs"
)
GAMMA = 0.1
SEED = 20260725
res = {"gamma": GAMMA, "seed": SEED, "theorem": "3.8"}

# ---------------------------------------------------------------------------
# (a) the two dimensions of the construction, computed exhaustively
# ---------------------------------------------------------------------------
print("== dimensions of the Theorem 3.8 class ==")
dims = []
for ku, d in ((4, 2), (5, 2), (6, 2), (6, 3), (8, 4)):
    cls, subsets = thm38_class(ku, d, GAMMA)
    dg = gamma_graph_dim(cls, GAMMA, max_d=min(ku, d + 2))
    max_k = min(5, ku)
    oig, oig_detail = gamma_oig_dim(cls, GAMMA, max_k=max_k)
    dims.append(
        {
            "ku": ku,
            "d": d,
            "n_hyp": cls.n_hyp,
            "d_gamma": dg,
            "d_gamma_expected": min(d, ku - d),
            "oig_dim": oig,
            "oig_le_3": bool(oig <= 3),
            "oig_detail": oig_detail,
            "oig_scanned_up_to_k": max_k,
        }
    )
    print(
        f"  ku={ku} d={d} |H|={cls.n_hyp:3d}:  d_gamma={dg} (expect {min(d,ku-d)})   "
        f"gamma-OIG dim={oig} (<=3: {oig<=3})   "
        f"min-max-outdeg per k={[ (k,v['min_max_outdegree']) for k,v in oig_detail.items() ]}"
    )
res["dimensions"] = dims
res["all_oig_le_3"] = bool(all(x["oig_le_3"] for x in dims))
res["all_dgamma_match"] = bool(all(x["d_gamma"] == x["d_gamma_expected"] for x in dims))

# rules used below really are interpolating (Definition 3.7)
rng = np.random.default_rng(SEED)
IRULES = {
    "mean": rule_mean,
    "median": rule_median,
    "midrange": rule_midrange,
    "min": rule_min,
    "max": rule_max,
}
res["rule_interpolating"] = {
    k: bool(is_interpolating_rule(v, rng)) for k, v in IRULES.items()
}
print("  Definition 3.7 check (interpolating?):", res["rule_interpolating"])

# an interpolating rule cannot rescue a point where every selected hypothesis
# is gamma-far: check explicitly on the actual value range of the construction
vals = np.array([GAMMA + 0.05 * GAMMA, GAMMA + 0.55 * GAMMA])
res["interpolating_rule_cannot_rescue"] = {
    k: bool(abs(f(vals) - 0.0) > GAMMA) for k, f in IRULES.items()
}
print(
    "  every interpolating rule still errs when all inputs are in (gamma,1]:",
    res["interpolating_rule_cannot_rescue"],
)


# ---------------------------------------------------------------------------
# (b) the hard distribution and the best-effort finite aggregation algorithm
# ---------------------------------------------------------------------------
def simulate_finite_aggregation(d, eps, n, m, ku, trials, seed, a_const=2.0):
    """A ~ uniform: A_1 = 1, A_2..A_d without replacement from {2,...,ku}.
    D_A: mass 1 - a*eps on A_1 and a*eps/(d-1) on each A_j, all labelled 0.
    The finite aggregation algorithm selects m hypotheses h_{B_1},...,h_{B_m}
    (|B_j| = d) -- best effort: cover every observed point, then spend the
    remaining m*d - |obs| slots on uniformly random unobserved points.
    Every point outside Z = union B_j is mispredicted by any interpolating rule.
    """
    rng = np.random.default_rng(seed)
    p = np.empty(d)
    p[0] = 1.0 - a_const * eps
    p[1:] = a_const * eps / (d - 1)
    losses = np.empty(trials)
    for t in range(trials):
        A = np.empty(d, dtype=np.int64)
        A[0] = 1
        rest = rng.integers(2, ku + 1, size=d - 1)  # ku >> d: collisions negligible
        while len(np.unique(rest)) != d - 1:
            rest = rng.integers(2, ku + 1, size=d - 1)
        A[1:] = rest
        draws = rng.choice(d, size=n, p=p)
        obs = np.unique(A[draws])
        budget = m * d
        n_guess = max(0, budget - len(obs))
        # guesses: uniform over the universe minus the observed points
        guesses = (
            rng.integers(1, ku + 1, size=n_guess) if n_guess else np.empty(0, np.int64)
        )
        Z = np.union1d(obs, guesses)
        missed = ~np.isin(A, Z)
        losses[t] = float(p[missed].sum())
    return float(losses.mean()), float(losses.std(ddof=1) / math.sqrt(trials))


print(
    "\n== Theorem 3.8: finite aggregation with an interpolating rule, n = floor(d/(128 eps)) =="
)
grid = []
TRIALS = 3000
for d in (32, 64, 128, 256):
    for eps in (0.1, 0.05, 0.02, 0.01):
        n = int(np.floor(d / (128 * eps)))
        for m in (1, 3, 10, 100):
            ku = max(1000, 1000 * d * m)  # "ku sufficiently large relative to d*m(n)"
            mean, se = simulate_finite_aggregation(
                d, eps, n, m, ku, TRIALS, SEED + d * 100 + m
            )
            grid.append(
                {
                    "d_gamma": d,
                    "eps": eps,
                    "n": n,
                    "m_hypotheses": m,
                    "ku": ku,
                    "E_loss": mean,
                    "stderr": se,
                    "E_loss_over_eps": mean / eps,
                    "holds": bool(mean - 2 * se > eps),
                }
            )
res["grid"] = grid
n_hold = sum(g["holds"] for g in grid)
res["grid_hold"] = f"{n_hold}/{len(grid)}"
res["min_ratio"] = min(g["E_loss_over_eps"] for g in grid)
print(
    f"  {n_hold}/{len(grid)} cells satisfy E[L] > eps (2-sigma);  min E[L]/eps = {res['min_ratio']:.3f}"
)
for g in grid:
    if g["eps"] == 0.01:
        print(
            f"    d={g['d_gamma']:3d} eps={g['eps']} n={g['n']:4d} m={g['m_hypotheses']:4d} ku={g['ku']:8d}"
            f"  E[L]={g['E_loss']:.5f} +/- {2*g['stderr']:.5f}   ({g['E_loss_over_eps']:.2f} x eps)"
        )

# how large must m be before the bound breaks?  (the theorem lets ku grow with m)
print(
    "\n  m-sweep at fixed ku (shows the bound is about ku >> d*m, not about m alone):"
)
msweep = []
d, eps = 128, 0.01
n = int(np.floor(d / (128 * eps)))
ku_fixed = 20000
for m in (1, 10, 100, 300, 1000):
    mean, se = simulate_finite_aggregation(d, eps, n, m, ku_fixed, 2000, SEED + 5 + m)
    msweep.append(
        {
            "m": m,
            "ku": ku_fixed,
            "d_times_m_over_ku": d * m / ku_fixed,
            "E_loss": mean,
            "E_loss_over_eps": mean / eps,
            "holds": bool(mean > eps),
        }
    )
    print(
        f"    m={m:5d}  d*m/ku={d*m/ku_fixed:7.3f}  E[L]/eps={mean/eps:6.3f} {'(holds)' if mean>eps else '(broken: universe too small)'}"
    )
res["m_sweep_fixed_ku"] = msweep


# ---------------------------------------------------------------------------
# (b2) where does the loss actually cross eps?  n*(eps) should be Theta(d/eps).
# ---------------------------------------------------------------------------
print("\n== empirical threshold n*(eps): smallest n with E[L] <= eps  (m=10) ==")
thr = []
for d in (32, 64, 128):
    for eps in (0.05, 0.02, 0.01):
        lo, hi = 1, 1 << 17
        while lo < hi:
            mid = (lo + hi) // 2
            mean, _ = simulate_finite_aggregation(
                d, eps, mid, 10, 1000 * d * 10, 800, SEED + 77
            )
            if mean <= eps:
                hi = mid
            else:
                lo = mid + 1
        thr.append(
            {"d_gamma": d, "eps": eps, "n_star": lo, "n_star_over_d_over_eps": lo / (d / eps)}
        )
        print(f"    d={d:4d} eps={eps:<5} n*={lo:7d}   n*/(d/eps)={lo/(d/eps):.4f}")
res["empirical_threshold"] = thr
rat = [t["n_star_over_d_over_eps"] for t in thr]
res["n_star_ratio_mean"] = float(np.mean(rat))
res["n_star_ratio_cv"] = float(np.std(rat) / np.mean(rat))
print(f"  n*/(d/eps): mean={np.mean(rat):.4f}  coeff-of-variation={np.std(rat)/np.mean(rat):.4f} "
      f"(constant => n* = Theta(d_gamma/eps))")

# ---------------------------------------------------------------------------
# (c) the separation: a general improper learner on the SAME class
# ---------------------------------------------------------------------------
print("\n== separation: a general (non-finite-aggregation) improper learner ==")


def improper_learner_loss(d, ku, n, trials, seed, p_nonzero):
    """Learner: if the sample ever shows a non-zero label it reveals the unique
    gamma_A, which identifies the target hypothesis exactly -> zero loss.
    Otherwise predict the constant 0 function (NOT in H, and not reachable by
    any interpolating aggregation of members of H).
    Distribution: mass p_nonzero spread on points where the target is non-zero,
    the rest on its zero set.  Loss = p_nonzero * 1{no non-zero label seen}."""
    rng = np.random.default_rng(seed)
    seen = rng.random((trials, n)) < p_nonzero
    identified = seen.any(axis=1)
    return float(np.mean(np.where(identified, 0.0, p_nonzero)))


sep = []
for d in (32, 64, 128, 256):
    eps = 0.01
    n = max(1, int(np.floor(d / (128 * eps))))
    # worst case over the free parameter p_nonzero
    worst = max(
        (improper_learner_loss(d, 10**6, n, 4000, SEED + 11, p), p)
        for p in np.linspace(0.005, 0.9, 60)
    )
    fa = [
        g
        for g in grid
        if g["d_gamma"] == d and g["eps"] == eps and g["m_hypotheses"] == 100
    ][0]
    sep.append(
        {
            "d_gamma": d,
            "eps": eps,
            "n": n,
            "finite_aggregation_E_loss": fa["E_loss"],
            "general_learner_worst_E_loss": worst[0],
            "general_learner_worst_p": float(worst[1]),
            "theory_bound_1_over_en": 1.0 / (math.e * n),
            "gap_factor": fa["E_loss"] / max(worst[0], 1e-12),
        }
    )
    print(
        f"  d_gamma={d:3d} n={n:4d}:  finite aggregation E[L]={fa['E_loss']:.5f}   "
        f"general learner worst-case E[L]={worst[0]:.5f}  (1/(en)={1/(math.e*n):.5f})   "
        f"gap = {sep[-1]['gap_factor']:.1f}x"
    )
res["separation"] = sep
res["gap_grows_with_dgamma"] = bool(sep[-1]["gap_factor"] > sep[0]["gap_factor"])

ok = (
    res["all_oig_le_3"]
    and res["all_dgamma_match"]
    and n_hold == len(grid)
    and res["gap_grows_with_dgamma"]
)
res["verdict"] = "verified" if ok else "partial"
print(f"\nverdict = {res['verdict']}")
dump_json(os.path.join(OUT, "claim3_thm38.json"), res)