scripts/claim6_pls_localmaxcut.py

"""Claim 6 -- Theorem 4.4 of arXiv:2601.20180.

"For the special case of strategic classification, Theorem 4.4 shows that computing
 a local optimum of the performative risk under single-label updates is PLS-hard,
 complementing prior NP-hardness results for global performative optimality."

PLS-hardness is inherited from the PLS-completeness of LocalMaxCut (Schaeffer &
Yannakakis 1991) and is not itself executable.  What IS executable is the reduction
of Appendix F, and we check it EXHAUSTIVELY (every one of the 2^|X| classifiers) on
small graphs:

  1. c is a metric;
  2. the Contestant's best response matches the paper's description;
  3. DIRECTION 1: every strategic local optimum labels all edge points 0 and its
     vertex labelling is a local max cut   <-- THIS FAILS, with a minimal
     counterexample on the single-edge graph;
  4. DIRECTION 2: every local max cut lifts to a strategic local optimum -- holds;
  5. the utility identity M*U(f) = w(cut) + const on cut-induced classifiers, and
     Eq. (19)-(20), (22), (24) -- all hold exactly;
  6. a minimal REPAIR (one extra "guard" point per edge) that restores direction 1,
     re-verified exhaustively;
  7. boundary audit of the two cost values.
"""

import itertools

import numpy as np

from common import dump

SEED = 20260725
rng = np.random.default_rng(SEED)
res = {"seed": SEED, "claim": "Theorem 4.4 (PLS-hardness via LocalMaxCut)"}

CLOSE, FAR = 0.8, 1.2


class Instance:
    """Strategic classification instance built from a weighted graph.

    guards=True adds the repair: for every edge one extra point with h=0 and
    weight 1 that is close to x_{(u,v)-} only.
    """

    def __init__(self, n, edges, weights, close=CLOSE, far=FAR, guards=False):
        self.n, self.edges, self.w = n, list(edges), dict(zip(edges, weights))
        self.close, self.far, self.guards = close, far, guards
        self.points = [("v", v) for v in range(n)]
        self.points += [("e+", e) for e in self.edges] + [("e-", e) for e in self.edges]
        if guards:
            self.points += [("g", e) for e in self.edges]
        self.idx = {p: i for i, p in enumerate(self.points)}
        self.N = len(self.points)
        self.h = np.array([1 if p[0] == "e+" else 0 for p in self.points], dtype=int)
        wD = []
        for p in self.points:
            if p[0] == "v":
                wD.append(sum(self.w[e] for e in self.edges if p[1] in e))
            elif p[0] == "e+":
                wD.append(2 * self.w[p[1]])
            elif p[0] == "e-":
                wD.append(2 * self.w[p[1]] + 1)
            else:
                wD.append(1.0)
        self.wD = np.array(wD, dtype=float)
        self.M = float(self.wD.sum())
        C = np.full((self.N, self.N), self.far)
        np.fill_diagonal(C, 0.0)
        for e in self.edges:
            ip, im = self.idx[("e+", e)], self.idx[("e-", e)]
            C[ip, im] = C[im, ip] = self.close
            for v in e:
                iv = self.idx[("v", v)]
                C[iv, ip] = C[ip, iv] = self.close
            if guards:
                ig = self.idx[("g", e)]
                C[ig, im] = C[im, ig] = self.close
        self.C = C
        # bitmask of "close" neighbours of each point (cost < 1, so deviation pays)
        self.close_mask = [
            sum(1 << j for j in range(self.N) if j != i and C[i, j] < 1.0)
            for i in range(self.N)
        ]

    # ------------------------------------------------------------- checks
    def metric_report(self):
        bad = 0
        for i, j, k in itertools.product(range(self.N), repeat=3):
            if self.C[i, k] > self.C[i, j] + self.C[j, k] + 1e-12:
                bad += 1
        return {
            "triangle_violations": bad,
            "symmetric": bool(np.allclose(self.C, self.C.T)),
            "zero_diagonal": bool(np.allclose(np.diag(self.C), 0)),
            "positive_off_diagonal": bool(
                np.all(self.C[~np.eye(self.N, dtype=bool)] > 0)
            ),
        }

    def all_utilities(self):
        """Vectorised utility of every one of the 2^N classifiers.

        Contestant best response: a point deviates iff it is labelled 0 and some
        close point is labelled 1 (staying pays f(x); moving pays f(y)-c(x,y)).
        So f(Delta(x)) = f(x) OR [exists close y : f(y)=1].
        """
        masks = np.arange(1 << self.N, dtype=np.int64)
        total = np.zeros(masks.shape, dtype=float)
        for i in range(self.N):
            fi = (masks >> i) & 1
            dev = ((masks & self.close_mask[i]) != 0).astype(np.int64)
            label_seen = np.maximum(fi, dev)  # f(Delta(x))
            correct = (label_seen == self.h[i]).astype(float)
            total += self.wD[i] * correct
        return total / self.M

    def local_optima(self, U):
        masks = np.arange(1 << self.N, dtype=np.int64)
        ok = np.ones(masks.shape, dtype=bool)
        for i in range(self.N):
            ok &= U >= U[masks ^ (1 << i)] - 1e-12
        return np.nonzero(ok)[0]

    def bits(self, mask):
        return np.array([(mask >> i) & 1 for i in range(self.N)], dtype=int)

    def vertex_labels(self, mask):
        return np.array(
            [(mask >> self.idx[("v", v)]) & 1 for v in range(self.n)], dtype=int
        )

    def edge_points_zero(self, mask):
        return all(
            ((mask >> self.idx[(t, e)]) & 1) == 0
            for e in self.edges
            for t in ("e+", "e-")
        )

    def cut_weight(self, S):
        return float(sum(self.w[e] for e in self.edges if S[e[0]] != S[e[1]]))

    def local_max_cuts(self):
        out = []
        for b in itertools.product([0, 1], repeat=self.n):
            S = np.array(b, dtype=int)
            base = self.cut_weight(S)
            good = True
            for v in range(self.n):
                T = S.copy()
                T[v] ^= 1
                if self.cut_weight(T) > base + 1e-12:
                    good = False
                    break
            if good:
                out.append(S)
        return out

    def cut_mask(self, S):
        m = 0
        for v in range(self.n):
            if S[v]:
                m |= 1 << self.idx[("v", v)]
        return m


def random_graph(n, m, rng):
    all_e = [(i, j) for i in range(n) for j in range(i + 1, n)]
    idx = rng.choice(len(all_e), size=min(m, len(all_e)), replace=False)
    edges = [all_e[i] for i in sorted(idx)]
    return edges, [float(rng.integers(1, 6)) for _ in edges]


def audit(n, edges, weights, guards):
    inst = Instance(n, edges, weights, guards=guards)
    U = inst.all_utilities()
    lo = inst.local_optima(U)
    lmc = inst.local_max_cuts()
    lmc_set = {tuple(s) for s in lmc}
    # direction 1
    bad = [
        int(m)
        for m in lo
        if not (inst.edge_points_zero(m) and tuple(inst.vertex_labels(m)) in lmc_set)
    ]
    # direction 2
    dir2 = all(int(inst.cut_mask(S)) in set(lo.tolist()) for S in lmc)
    # utility identity on cut-induced classifiers
    consts = [
        inst.M * U[inst.cut_mask(np.array(b))] - inst.cut_weight(np.array(b))
        for b in itertools.product([0, 1], repeat=n)
    ]
    identity = float(np.ptp(np.array(consts)))
    # global optima are max cuts
    best = U.max()
    maxcut = max(
        inst.cut_weight(np.array(b)) for b in itertools.product([0, 1], repeat=n)
    )
    gopt = np.nonzero(U >= best - 1e-12)[0]
    global_ok = all(
        inst.edge_points_zero(int(m))
        and inst.cut_weight(inst.vertex_labels(int(m))) >= maxcut - 1e-12
        for m in gopt
    )
    return (
        {
            "n_vertices": n,
            "n_edges": len(edges),
            "population_size": inst.N,
            "classifiers_enumerated": 1 << inst.N,
            "guards": guards,
            "n_strategic_local_optima": int(len(lo)),
            "n_local_max_cuts": len(lmc),
            "n_local_optima_violating_direction1": len(bad),
            "direction1_every_local_opt_is_a_local_max_cut": bool(len(bad) == 0),
            "direction2_every_local_max_cut_is_a_local_opt": bool(dir2),
            "utility_identity_spread": identity,
            "global_optima_are_max_cuts": bool(global_ok),
            "max_cut_weight": float(maxcut),
            "worst_spurious_utility_times_M": (
                float(min(U[m] for m in bad) * inst.M) if bad else None
            ),
            "cut_induced_utility_times_M_range": [
                float(min(consts)),
                float(max(consts)),
            ],
        },
        inst,
        U,
        lo,
        bad,
    )


graphs = [(2, [(0, 1)], [1.0])]
for n, m in [(3, 2), (3, 3), (4, 3), (4, 4), (5, 4)]:
    for _ in range(2):
        e, w = random_graph(n, m, rng)
        graphs.append((n, e, w))

orig, repaired = [], []
print("=== ORIGINAL construction (Appendix F as published) ===")
for n, edges, weights in graphs:
    r, inst, U, lo, bad = audit(n, edges, weights, guards=False)
    orig.append(r)
    print(
        "  n=%d m=%d |X|=%2d  %7d classifiers | %4d local optima, %2d local max cuts "
        "| direction1 %s (%d violations) | direction2 %s"
        % (
            n,
            len(edges),
            r["population_size"],
            r["classifiers_enumerated"],
            r["n_strategic_local_optima"],
            r["n_local_max_cuts"],
            r["direction1_every_local_opt_is_a_local_max_cut"],
            r["n_local_optima_violating_direction1"],
            r["direction2_every_local_max_cut_is_a_local_opt"],
        )
    )

print("=== REPAIRED construction (one guard point per edge) ===")
for n, edges, weights in graphs:
    r, inst, U, lo, bad = audit(n, edges, weights, guards=True)
    repaired.append(r)
    print(
        "  n=%d m=%d |X|=%2d  %7d classifiers | %4d local optima, %2d local max cuts "
        "| direction1 %s (%d violations) | direction2 %s"
        % (
            n,
            len(edges),
            r["population_size"],
            r["classifiers_enumerated"],
            r["n_strategic_local_optima"],
            r["n_local_max_cuts"],
            r["direction1_every_local_opt_is_a_local_max_cut"],
            r["n_local_optima_violating_direction1"],
            r["direction2_every_local_max_cut_is_a_local_opt"],
        )
    )

res["original_construction"] = orig
res["repaired_construction"] = repaired

# ------------------------------------------------ the minimal counterexample
inst = Instance(2, [(0, 1)], [1.0], guards=False)
U = inst.all_utilities()
lo = inst.local_optima(U)
lmc_set = {tuple(s) for s in inst.local_max_cuts()}
ce = None
for m in lo:
    m = int(m)
    if not (inst.edge_points_zero(m) and tuple(inst.vertex_labels(m)) in lmc_set):
        ce = m
        break
detail = {
    "graph": "single edge (0,1) with w = 1",
    "population": [str(p) for p in inst.points],
    "h": inst.h.tolist(),
    "weights_wD": inst.wD.tolist(),
    "M": inst.M,
    "counterexample_labels": {
        str(inst.points[i]): int((ce >> i) & 1) for i in range(inst.N)
    },
    "counterexample_utility_times_M": float(U[ce] * inst.M),
    "single_flip_utilities_times_M": {
        str(inst.points[i]): float(U[ce ^ (1 << i)] * inst.M) for i in range(inst.N)
    },
    "best_utility_times_M": float(U.max() * inst.M),
    "induced_vertex_labels": inst.vertex_labels(ce).tolist(),
    "induced_cut_weight": inst.cut_weight(inst.vertex_labels(ce)),
    "max_cut_weight": 1.0,
    "induced_cut_is_local_max": bool(tuple(inst.vertex_labels(ce)) in lmc_set),
    "why": (
        "The proof's first claim says a Jury with f(x_{(u,v)-}) = 1 gains at least "
        "1/M by flipping it to 0, because x_{(u,v)-} then becomes correctly "
        "classified.  That step silently assumes f(x_{(u,v)+}) = 0.  When "
        "f(x_{(u,v)+}) = 1 the point x_{(u,v)-} strategically deviates to "
        "x_{(u,v)+} after the flip and is still misclassified, so the gain is 0 "
        "and the configuration is a strategic local optimum with edge points "
        "labelled 1."
    ),
}
res["minimal_counterexample"] = detail
print("\nMinimal counterexample (single edge, w=1):")
print("  labels:", detail["counterexample_labels"])
print(
    "  M*U = %.1f ; single-flip M*U = %s ; best M*U = %.1f"
    % (
        detail["counterexample_utility_times_M"],
        detail["single_flip_utilities_times_M"],
        detail["best_utility_times_M"],
    )
)
print(
    "  induced vertex labels %s -> cut weight %.1f (max cut %.1f), local max cut: %s"
    % (
        detail["induced_vertex_labels"],
        detail["induced_cut_weight"],
        detail["max_cut_weight"],
        detail["induced_cut_is_local_max"],
    )
)

res["metric_check"] = Instance(
    4, [(0, 1), (1, 2), (2, 3), (0, 3)], [1.0, 2.0, 3.0, 1.0]
).metric_report()
res["metric_check_repaired"] = Instance(
    4, [(0, 1), (1, 2), (2, 3), (0, 3)], [1.0, 2.0, 3.0, 1.0], guards=True
).metric_report()
print(
    "\nmetric (original):",
    res["metric_check"],
    "| (repaired):",
    res["metric_check_repaired"],
)

# --------------------------------------------------------- cost boundary audit
cost_audit = []
for close, far in [
    (0.8, 1.2),
    (0.9, 1.1),
    (0.5, 1.2),
    (0.5, 1.5),
    (0.1, 5.0),
    (0.99, 1.01),
]:
    n, edges, weights = 4, [(0, 1), (1, 2), (2, 3), (0, 3)], [1.0, 2.0, 3.0, 1.0]
    i0 = Instance(n, edges, weights, close=close, far=far, guards=True)
    m = i0.metric_report()
    U = i0.all_utilities()
    lo = i0.local_optima(U)
    lmc_set = {tuple(s) for s in i0.local_max_cuts()}
    ok = all(
        i0.edge_points_zero(int(x)) and tuple(i0.vertex_labels(int(x))) in lmc_set
        for x in lo
    )
    cost_audit.append(
        {
            "c_close": close,
            "c_far": far,
            "is_metric": bool(m["triangle_violations"] == 0),
            "triangle_violations": m["triangle_violations"],
            "far_le_2_close": bool(far <= 2 * close + 1e-12),
            "repaired_reduction_correct": bool(ok),
        }
    )
res["cost_boundary_audit"] = cost_audit
print(
    "\ncost audit (the paper says the two values may be 'arbitrary' in (0,1) and (1,inf)):"
)
for a in cost_audit:
    print(
        "  close=%.2f far=%.2f : metric=%-5s (violations %3d; far<=2*close: %-5s) | "
        "repaired reduction correct=%s"
        % (
            a["c_close"],
            a["c_far"],
            a["is_metric"],
            a["triangle_violations"],
            a["far_le_2_close"],
            a["repaired_reduction_correct"],
        )
    )

res["totals"] = {
    "classifiers_enumerated_total": int(
        sum(r["classifiers_enumerated"] for r in orig + repaired)
    ),
    "original_direction1_holds_everywhere": bool(
        all(r["direction1_every_local_opt_is_a_local_max_cut"] for r in orig)
    ),
    "original_direction2_holds_everywhere": bool(
        all(r["direction2_every_local_max_cut_is_a_local_opt"] for r in orig)
    ),
    "repaired_direction1_holds_everywhere": bool(
        all(r["direction1_every_local_opt_is_a_local_max_cut"] for r in repaired)
    ),
    "repaired_direction2_holds_everywhere": bool(
        all(r["direction2_every_local_max_cut_is_a_local_opt"] for r in repaired)
    ),
    "utility_identity_max_spread": float(
        max(r["utility_identity_spread"] for r in orig + repaired)
    ),
    "global_optima_are_max_cuts_repaired": bool(
        all(r["global_optima_are_max_cuts"] for r in repaired)
    ),
}
print("\nTOTALS:", res["totals"])

res["verdict"] = {
    "published_reduction_is_solution_preserving": res["totals"][
        "original_direction1_holds_everywhere"
    ],
    "repaired_reduction_is_solution_preserving": res["totals"][
        "repaired_direction1_holds_everywhere"
    ],
    "summary": (
        "The published reduction (Appendix F) admits strategic local optima whose "
        "vertex labelling is not a local max cut, so as written it is not a valid "
        "PLS reduction: the first claim of the proof fails whenever an edge's "
        "positive and negative points are both labelled 1.  Adding one guard point "
        "per edge (h=0, weight 1, close only to x_{(u,v)-}) removes every spurious "
        "local optimum and restores the reduction; the theorem's conclusion "
        "(PLS-hardness) therefore stands."
    ),
}
dump("claim6_pls_localmaxcut.json", res)