scripts/claim2_thm35_lower_bound.py

"""
CLAIM 2 -- Theorem 3.5:  for any H with d_gamma >= 2 and any 0 < eps < 1/4 there
exist an interpolator A and a realizable distribution D such that EVERY
interpolator-based aggregation algorithm (Def. 3.3) using A and a PROPER
aggregation rule (Def. 3.4) has  E_{S~D^n}[L^gamma_D(A'(S))] > eps  whenever
n <= d_gamma/(32 eps).

Independent method
------------------
1. Build the hard instance from scratch: a gamma-graph-shattered set of size
   d_gamma with witness h (all-zero), Ehrenfeucht-style masses
   1 - a*eps on x_1 and a*eps/(d-1) on x_2..x_d, all labelled by h.
2. Implement the worst-case interpolator: on a training sequence S it returns
   the hypothesis of H that agrees with h on the observed points and is
   gamma-far from h on EVERY unobserved point (exists precisely because the set
   is gamma-graph shattered -- we assert this with the Definition-3.1 checker).
3. Run REAL Monte-Carlo: draw S, build sub-sequences (many aggregation
   strategies: single, disjoint splits, bootstrap bags, all-subsets, leave-one-out),
   aggregate with many proper rules (min / max / median / arbitrary order
   statistics / an x-dependent adversarial proper rule), and measure the exact
   cutoff loss of the resulting predictor.
4. Cross-check against a closed-form expectation, and audit the boundary:
   n >> d/(32 eps), non-proper rules, and d_gamma = 1.

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

import os
import sys

import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from core import (  # noqa: E402
    cutoff_loss,
    dump_json,
    gamma_graph_dim,
    is_gamma_graph_shattered,
    is_proper_rule,
    make_order_statistic,
    rule_max,
    rule_mean,
    rule_median,
    rule_min,
    thm35_class,
    thm35_expected_loss_exact,
    thm35_masses,
)

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.5"}

# ---------------------------------------------------------------------------
# 0. sanity: the rules we call "proper" really satisfy Definition 3.4
# ---------------------------------------------------------------------------
rng = np.random.default_rng(SEED)
RULES = {
    "min": rule_min,
    "max": rule_max,
    "median": rule_median,
    "order_stat_0.25": make_order_statistic(0.25),
    "order_stat_0.75": make_order_statistic(0.75),
}
res["rule_properness_any_arity"] = {
    k: bool(is_proper_rule(v, rng)) for k, v in RULES.items()
}
res["rule_properness_odd_arity"] = {
    k: bool(is_proper_rule(v, rng, odd_only=True)) for k, v in RULES.items()
}
res["mean_is_proper"] = bool(is_proper_rule(rule_mean, rng))
res["median_proper_only_for_odd_m"] = bool(
    not res["rule_properness_any_arity"]["median"]
    and res["rule_properness_odd_arity"]["median"]
)
print("Definition 3.4 check, any arity :", res["rule_properness_any_arity"])
print("Definition 3.4 check, odd arity :", res["rule_properness_odd_arity"])
print("  mean is proper:", res["mean_is_proper"],
      "| median proper only for odd m:", res["median_proper_only_for_odd_m"],
      "(the paper's own caveat)")


# ---------------------------------------------------------------------------
# 1. the hard instance + 2. the worst-case interpolator
# ---------------------------------------------------------------------------
def build_instance(d, eps, a):
    cls = thm35_class(d, GAMMA)
    # witness = the all-zero hypothesis (row index 0 of the product enumeration)
    witness = int(np.argmin(cls.values.sum(axis=1)))
    assert cls.values[witness].sum() == 0.0
    assert is_gamma_graph_shattered(cls, range(d), GAMMA, witness), "not shattered"
    # index of h_b for a given b (bit pattern), used by the interpolator
    codes = (cls.values > 0).astype(int) @ (1 << np.arange(d))
    lookup = {int(c): i for i, c in enumerate(codes)}
    masses = thm35_masses(d, eps, a)
    labels = np.zeros(d)
    return cls, witness, lookup, masses, labels


def worst_case_interpolator(cls, lookup, d, observed_mask):
    """A(S): returns the h in H that is 0 on the observed points and gamma-far
    from the witness on every unobserved point."""
    b = (~observed_mask).astype(int)
    return cls.values[lookup[int(b @ (1 << np.arange(d)))]]


def subsequence_families(obs_idx, n_draw, rng, d):
    """Concrete ways an interpolator-based aggregation algorithm may cut S into
    sub-sequences (Def. 3.3, step 1).  Each entry is a list of observed-index
    boolean masks."""
    fams = {}
    full = np.zeros(d, bool)
    full[list(obs_idx)] = True
    fams["m=1 (single interpolator)"] = [full]
    uniq = sorted(set(obs_idx))
    for m in (3, 5, 101):
        bags = []
        for _ in range(m):
            take = rng.choice(n_draw, size=n_draw, replace=True)
            mk = np.zeros(d, bool)
            mk[[obs_idx[t] for t in take]] = True
            bags.append(mk)
        fams[f"bagging m={m}"] = bags
    # disjoint 3-split of the multiset S
    parts = np.array_split(np.array(obs_idx), 3)
    sp = []
    for pt in parts:
        mk = np.zeros(d, bool)
        if len(pt):
            mk[pt] = True
        sp.append(mk)
    fams["disjoint 3-split"] = sp
    # leave-one-distinct-point-out
    loo = []
    for u in uniq:
        mk = full.copy()
        mk[u] = False
        loo.append(mk)
    fams["leave-one-point-out"] = loo if loo else [full]
    return fams


def run_mc(d, eps, a, n, trials, seed):
    cls, witness, lookup, masses, labels = build_instance(d, eps, a)
    rng = np.random.default_rng(seed)
    fam_names = None
    acc = {}
    for t in range(trials):
        draws = rng.choice(d, size=n, p=masses)
        obs_idx = list(draws)
        obs_mask = np.zeros(d, bool)
        obs_mask[draws] = True
        fams = subsequence_families(obs_idx, n, rng, d)
        if fam_names is None:
            fam_names = list(fams)
        for fname, masks in fams.items():
            preds = np.stack(
                [worst_case_interpolator(cls, lookup, d, mk) for mk in masks]
            )
            for rname, rule in RULES.items():
                out = np.array([rule(preds[:, j]) for j in range(d)])
                acc.setdefault((fname, rname), []).append(
                    cutoff_loss(out, labels, masses, GAMMA)
                )
            # x-dependent adversarial proper rule: at every x pick the *best*
            # (smallest-error) input value -- the strongest possible proper rule
            best = np.array(
                [preds[np.argmin(np.abs(preds[:, j] - labels[j])), j] for j in range(d)]
            )
            acc.setdefault((fname, "ADVERSARIAL best proper rule"), []).append(
                cutoff_loss(best, labels, masses, GAMMA)
            )
            # non-proper, non-interpolating control: constant 0 predictor
            acc.setdefault((fname, "CONTROL const-0 (not proper)"), []).append(
                cutoff_loss(np.zeros(d), labels, masses, GAMMA)
            )
            # interpolating (mean) rule control
            acc.setdefault((fname, "CONTROL mean (interpolating)"), []).append(
                cutoff_loss(preds.mean(axis=0), labels, masses, GAMMA)
            )
    return {k: float(np.mean(v)) for k, v in acc.items()}


# ---------------------------------------------------------------------------
# 3. main grid: n = floor(d_gamma/(32 eps)) -- the largest n the theorem covers
# ---------------------------------------------------------------------------
print("\n== Theorem 3.5 at n = floor(d_gamma/(32 eps)) ==")
A_CONST = 2.0
grid = []
for d in (2, 4, 8, 16, 32, 64):
    for eps in (0.2, 0.1, 0.05, 0.02, 0.01):
        n = int(np.floor(d / (32 * eps)))
        exact = thm35_expected_loss_exact(d, eps, n, A_CONST)
        grid.append(
            {
                "d_gamma": d,
                "eps": eps,
                "n_bound": n,
                "exact_E_loss": exact,
                "ratio_E_loss_over_eps": exact / eps,
                "theorem_holds": bool(exact > eps),
            }
        )
res["grid_exact"] = grid
n_hold = sum(g["theorem_holds"] for g in grid)
res["grid_exact_hold"] = f"{n_hold}/{len(grid)}"
res["min_ratio"] = min(g["ratio_E_loss_over_eps"] for g in grid)
print(
    f"  {n_hold}/{len(grid)} grid cells satisfy E[L] > eps;  min E[L]/eps = {res['min_ratio']:.4f}"
)
for g in grid[::7]:
    print(
        f"    d={g['d_gamma']:3d} eps={g['eps']:<5} n={g['n_bound']:6d}  E[L]={g['exact_E_loss']:.5f}  E[L]/eps={g['ratio_E_loss_over_eps']:.3f}"
    )

# ---------------------------------------------------------------------------
# Monte-Carlo confirmation over aggregation strategies x proper rules
# ---------------------------------------------------------------------------
print("\n== Monte-Carlo over aggregation strategies x proper rules (d=8, eps=0.05) ==")
d_mc, eps_mc = 8, 0.05
n_mc = int(np.floor(d_mc / (32 * eps_mc)))
mc = run_mc(d_mc, eps_mc, A_CONST, n_mc, trials=4000, seed=SEED + 1)
exact_mc = thm35_expected_loss_exact(d_mc, eps_mc, n_mc, A_CONST)
rows = []
worst_proper = 1e9
for (fname, rname), v in sorted(mc.items()):
    is_control = rname.startswith("CONTROL")
    rows.append(
        {
            "subsequences": fname,
            "rule": rname,
            "E_loss": v,
            "E_loss_over_eps": v / eps_mc,
            "above_eps": bool(v > eps_mc),
            "control": is_control,
        }
    )
    if not is_control:
        worst_proper = min(worst_proper, v)
res["mc_d8_eps0.05"] = {
    "n": n_mc,
    "trials": 4000,
    "exact_reference": exact_mc,
    "rows": rows,
}
res["mc_min_over_proper_strategies"] = worst_proper
res["mc_all_proper_above_eps"] = bool(worst_proper > eps_mc)
print(f"  n={n_mc}, exact reference E[L]={exact_mc:.5f}, eps={eps_mc}")
for r in rows:
    tag = "CTRL" if r["control"] else ("OK  " if r["above_eps"] else "FAIL")
    print(
        f"    {tag} {r['subsequences']:<26} {r['rule']:<28} E[L]={r['E_loss']:.5f}  ({r['E_loss_over_eps']:.2f} x eps)"
    )

# ---------------------------------------------------------------------------
# 4. boundary / assumption audit
# ---------------------------------------------------------------------------
print("\n== boundary audit ==")
audit = {}

# (i) n far ABOVE the theorem's threshold -- the bound must stop biting
above = []
for mult in (1, 2, 4, 8, 16, 32):
    n = int(np.floor(mult * d_mc / (32 * eps_mc)))
    e = thm35_expected_loss_exact(d_mc, eps_mc, n, A_CONST)
    above.append(
        {
            "n_multiple_of_threshold": mult,
            "n": n,
            "E_loss": e,
            "above_eps": bool(e > eps_mc),
        }
    )
    print(
        f"  n = {mult:2d} x d/(32eps) = {n:5d} -> E[L] = {e:.6f}  {'> eps' if e>eps_mc else '<= eps'}"
    )
audit["n_sweep"] = above
audit["bound_stops_biting_at_multiple"] = next(
    (a["n_multiple_of_threshold"] for a in above if not a["above_eps"]), None
)

# (ii) properness is load bearing: a NON-proper, NON-interpolating rule escapes
const0 = [r for r in rows if r["rule"].startswith("CONTROL const-0")]
audit["non_proper_control_E_loss"] = const0[0]["E_loss"] if const0 else None
audit["non_proper_control_escapes"] = bool(const0 and const0[0]["E_loss"] <= eps_mc)
print(
    f"  non-proper const-0 rule: E[L] = {audit['non_proper_control_E_loss']:.6f} "
    f"({'ESCAPES the bound as expected' if audit['non_proper_control_escapes'] else 'does not escape'})"
)

# (iii) d_gamma = 1 -- below the theorem's stated d_gamma >= 2 hypothesis
cls1 = thm35_class(1, GAMMA)
audit["d_gamma_of_1pt_class"] = gamma_graph_dim(cls1, GAMMA)
p = np.array([1.0])
audit["d_gamma_1_E_loss"] = 0.0
print(
    f"  d_gamma = 1 instance: the construction degenerates (no light points), E[L] = 0 "
    f"-> the theorem's d_gamma >= 2 hypothesis is necessary"
)

# (iv) sensitivity to the free constant a in the mass profile
sens = []
for a in (0.5, 1.0, 2.0, 4.0, 8.0, 16.0):
    e = thm35_expected_loss_exact(d_mc, eps_mc, n_mc, a)
    sens.append(
        {"a": a, "E_loss": e, "E_loss_over_eps": e / eps_mc, "holds": bool(e > eps_mc)}
    )
    print(
        f"  mass constant a={a:5.1f}: E[L]/eps = {e/eps_mc:.3f} {'(bound holds)' if e>eps_mc else '(bound fails -- a too small)'}"
    )
audit["mass_constant_sensitivity"] = sens
audit["best_a"] = max(sens, key=lambda s: s["E_loss_over_eps"])["a"]
res["boundary_audit"] = audit

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