code/remaining_claim_routes.py

#!/usr/bin/env python3
"""Four independent routes for Claims 2, 4, 5, and 6.

The finite experiments simulate the distribution produced by the paper's
quantum sampling stage. They do not simulate the quantum leverage-score or
QMLSO subroutines and therefore cannot establish their runtimes.
"""
from __future__ import annotations

import json
import math
import os
import platform
import time
from pathlib import Path

import numpy as np


SEEDS = tuple(range(20))
HORIZONS = (8, 16, 32, 64, 128, 256, 512)
SOURCE_SHA = "bd48105ab08395ba1edbdb3a407eee9f2e1a8464521d7d67dbe5b6e96edf2549"


def cpu_count() -> int:
    if hasattr(os, "sched_getaffinity"):
        return len(os.sched_getaffinity(0))
    return os.cpu_count() or 1


def leverage_probabilities(a: np.ndarray) -> np.ndarray:
    leverage = np.einsum("ij,jk,ik->i", a, np.linalg.pinv(a.T @ a), a)
    leverage = np.maximum(leverage, 0)
    return leverage / leverage.sum()


def sampled_gram(a: np.ndarray, probabilities: np.ndarray, k: int, seed: int) -> np.ndarray:
    rng = np.random.default_rng(seed)
    indices = rng.choice(len(a), size=k, replace=True, p=probabilities)
    weights = np.bincount(indices, minlength=len(a)) / (k * probabilities)
    return a.T @ (weights[:, None] * a)


def embedding_error(a: np.ndarray, gram_sample: np.ndarray) -> float:
    gram = a.T @ a
    values, vectors = np.linalg.eigh(gram)
    positive = values > values.max() * 1e-12
    inverse_root = (vectors[:, positive] / np.sqrt(values[positive])) @ vectors[:, positive].T
    normalized = inverse_root @ gram_sample @ inverse_root
    projected_identity = vectors[:, positive] @ vectors[:, positive].T
    return float(np.linalg.norm(normalized - projected_identity, ord=2))


def hard_design(m: int, n: int, seed: int) -> tuple[np.ndarray, np.ndarray]:
    rng = np.random.default_rng(seed)
    a = rng.normal(scale=0.03, size=(m, n))
    a[:n] = np.eye(n) * math.sqrt(m)
    x_true = np.linspace(-0.7, 0.9, n)
    b = a @ x_true + rng.normal(scale=0.01, size=m)
    return a, b


def solve_sampled_regression(
    a: np.ndarray, b: np.ndarray, probabilities: np.ndarray, k: int, seed: int
) -> dict[str, float]:
    rng = np.random.default_rng(seed)
    indices = rng.choice(len(a), size=k, replace=True, p=probabilities)
    scale = np.sqrt(1 / (k * probabilities[indices]))
    sampled_a = a[indices] * scale[:, None]
    sampled_b = b[indices] * scale
    x_sampled = np.linalg.lstsq(sampled_a, sampled_b, rcond=None)[0]
    x_full = np.linalg.lstsq(a, b, rcond=None)[0]
    sampled_obj = float(np.sum((a @ x_sampled - b) ** 2))
    optimum = float(np.sum((a @ x_full - b) ** 2))
    return {
        "full_objective": optimum,
        "sampled_solution_objective": sampled_obj,
        "objective_ratio": sampled_obj / optimum,
    }


def grid_losses(b: np.ndarray, grid: np.ndarray, loss: str, p: float = 1.0) -> np.ndarray:
    residual = grid[None, :] - b[:, None]
    if loss == "huber":
        absolute = np.abs(residual)
        return np.where(absolute <= 1, 0.5 * residual**2, absolute - 0.5)
    return np.abs(residual) ** p


def grid_probabilities(losses: np.ndarray) -> np.ndarray:
    total = losses.sum(axis=0)
    ratios = np.divide(losses, total, out=np.zeros_like(losses), where=total > 1e-15)
    sensitivity = ratios.max(axis=1)
    sensitivity = np.maximum(sensitivity, 1e-15)
    return sensitivity / sensitivity.sum()


def grid_coreset_error(losses: np.ndarray, probabilities: np.ndarray, k: int, seed: int) -> float:
    rng = np.random.default_rng(seed)
    indices = rng.choice(len(losses), size=k, replace=True, p=probabilities)
    weights = np.bincount(indices, minlength=len(losses)) / (k * probabilities)
    full = losses.sum(axis=0)
    sampled = (weights[:, None] * losses).sum(axis=0)
    relative = np.abs(sampled - full) / np.maximum(full, 1e-12)
    return float(relative.max())


def first_hit(errors: dict[int, list[float]], tolerance: float, target_rate: float = 0.8) -> int | None:
    for horizon in HORIZONS:
        success_rate = sum(error <= tolerance for error in errors[horizon]) / len(errors[horizon])
        if success_rate >= target_rate:
            return horizon
    return None


def linear_sweep(ridge_lambda: float | None = None) -> dict:
    sizes = (512, 2048, 8192)
    cells = []
    control_failed = False
    for m in sizes:
        a, _ = hard_design(m, 8, 71 + m)
        if ridge_lambda is not None:
            a = np.vstack((a, math.sqrt(ridge_lambda) * np.eye(a.shape[1])))
        leverage = leverage_probabilities(a)
        uniform = np.full(len(a), 1 / len(a))
        errors = {k: [] for k in HORIZONS}
        control_errors = {k: [] for k in HORIZONS}
        for seed in SEEDS:
            for k in HORIZONS:
                errors[k].append(embedding_error(a, sampled_gram(a, leverage, k, seed)))
                control_errors[k].append(embedding_error(a, sampled_gram(a, uniform, k, seed)))
        hit = first_hit(errors, 0.5)
        control_hit = first_hit(control_errors, 0.5)
        if control_hit is None or (hit is not None and control_hit > hit):
            control_failed = True
        cells.append({
            "m": m,
            "first_hit_k": hit,
            "uniform_control_first_hit_k": control_hit,
            "success_rate_at_512": sum(x <= 0.5 for x in errors[512]) / len(SEEDS),
            "uniform_success_rate_at_512": sum(x <= 0.5 for x in control_errors[512]) / len(SEEDS),
        })
    return {
        "tolerance": 0.5,
        "success_rate_target": 0.8,
        "seeds": list(SEEDS),
        "horizons": list(HORIZONS),
        "cells": cells,
        "negative_control_failed_as_intended": control_failed,
    }


def loss_sweep(loss: str, p: float = 1.0, control: str = "uniform") -> dict:
    cells = []
    control_failed = False
    grid = np.linspace(-7, 7, 801)
    for m in (512, 2048, 8192):
        rng = np.random.default_rng(991 + m)
        b = rng.normal(scale=0.3, size=m)
        b[:8] = np.array([-6, -5, -4, -3, 3, 4, 5, 6])
        losses = grid_losses(b, grid, loss, p)
        informed = grid_probabilities(losses)
        uniform = np.full(m, 1 / m)
        errors = {k: [] for k in HORIZONS}
        control_errors = {k: [] for k in HORIZONS}
        full_curve = losses.sum(axis=0)
        single_row_error = float(
            np.max(np.abs(m * losses[0] - full_curve) / np.maximum(full_curve, 1e-12))
        )
        for seed in SEEDS:
            for k in HORIZONS:
                errors[k].append(grid_coreset_error(losses, informed, k, seed))
                if control == "single-row-support":
                    control_errors[k].append(single_row_error)
                else:
                    control_errors[k].append(grid_coreset_error(losses, uniform, k, seed))
        hit = first_hit(errors, 0.5)
        control_hit = first_hit(control_errors, 0.5)
        if control_hit is None or (hit is not None and control_hit > hit):
            control_failed = True
        cells.append({
            "m": m,
            "first_hit_k": hit,
            "uniform_control_first_hit_k": control_hit,
            "success_rate_at_512": sum(x <= 0.5 for x in errors[512]) / len(SEEDS),
            "uniform_success_rate_at_512": sum(x <= 0.5 for x in control_errors[512]) / len(SEEDS),
        })
    return {
        "loss": loss,
        "p": p,
        "negative_control": control,
        "grid_points": len(grid),
        "tolerance": 0.5,
        "success_rate_target": 0.8,
        "seeds": list(SEEDS),
        "horizons": list(HORIZONS),
        "cells": cells,
        "negative_control_failed_as_intended": control_failed,
    }


def finite_simulations() -> dict:
    a, b = hard_design(2048, 8, 2026)
    leverage = leverage_probabilities(a)
    linear = solve_sampled_regression(a, b, leverage, 256, 17)
    ridge_lambda = 0.5
    ridge_a = np.vstack((a, math.sqrt(ridge_lambda) * np.eye(8)))
    ridge_b = np.concatenate((b, np.zeros(8)))
    ridge = solve_sampled_regression(ridge_a, ridge_b, leverage_probabilities(ridge_a), 256, 17)

    grid = np.linspace(-7, 7, 1401)
    rng = np.random.default_rng(2026)
    targets = rng.normal(scale=0.5, size=2048)
    targets[:8] = np.array([-6, -5, -4, -3, 3, 4, 5, 6])
    loss_results = {}
    for name, p in (("huber", 1.0), ("lp_p_0.5", 0.5), ("lp_p_1.5", 1.5)):
        loss_kind = "huber" if name == "huber" else "lp"
        losses = grid_losses(targets, grid, loss_kind, p)
        probabilities = grid_probabilities(losses)
        rng_sample = np.random.default_rng(17)
        indices = rng_sample.choice(len(losses), size=256, replace=True, p=probabilities)
        weights = np.bincount(indices, minlength=len(losses)) / (256 * probabilities)
        full_curve = losses.sum(axis=0)
        sampled_curve = (weights[:, None] * losses).sum(axis=0)
        full_index = int(np.argmin(full_curve))
        sampled_index = int(np.argmin(sampled_curve))
        loss_results[name] = {
            "full_solution": float(grid[full_index]),
            "sampled_solution": float(grid[sampled_index]),
            "full_optimum": float(full_curve[full_index]),
            "sampled_solution_full_objective": float(full_curve[sampled_index]),
            "objective_ratio": float(full_curve[sampled_index] / full_curve[full_index]),
        }
    return {
        "linear": linear,
        "ridge": ridge,
        **loss_results,
        "scope": "finite classical simulation of the target sampling distribution; no quantum subroutine runtime is measured",
    }


def source_routes() -> dict:
    common = {
        "paper_source_sha256": SOURCE_SHA,
        "quantifiers": "algorithmic existence and high-probability guarantee over all inputs satisfying oracle assumptions",
        "certificate_found": False,
    }
    return {
        "C2": {**common, "anchor": "Corollary 23, source lines 1150-1153", "chain": "Theorem 19 quantum leverage scores -> sampling -> classical linear solve"},
        "C4": {**common, "anchor": "Corollary 25, source lines 1160-1163", "chain": "[A;sqrt(lambda)I] reduction -> Corollary 23"},
        "C5": {**common, "anchor": "gamma_p lines 263-270; Corollary 12 lines 552-554", "chain": "proper-loss specialization -> QGLMSparsify -> GLM solver oracle"},
        "C6": {**common, "anchor": "Corollary 11, source lines 549-551", "chain": "ell_p proper-loss specialization -> QGLMSparsify -> GLM solver oracle", "primary_reference_gap": "arXiv:2311.18145 states its regression application for p in (1,2], while the corollary claims p in (0,2]"},
    }


def falsification_routes() -> dict:
    # Exhaustive finite-domain searches cannot prove a universal runtime theorem,
    # but they can produce a valid counterexample if a displayed guarantee is
    # already impossible on a source-valid instance.
    scalar_values = (-2.0, -1.0, 0.0, 1.0, 2.0)
    finite_cases = 0
    impossible_guarantees = 0
    for a in scalar_values:
        for b in scalar_values:
            if a == 0:
                continue
            finite_cases += 1
            optimum = 0.0
            candidate = (a * (b / a) - b) ** 2
            impossible_guarantees += candidate > 1.1 * optimum + 1e-12

    properness_checks = []
    for p in (0.25, 0.5, 1.0, 1.5, 2.0):
        exponent = p / 2
        violation = 0.0
        for x in scalar_values:
            for y in scalar_values:
                violation = max(violation, abs(x + y) ** exponent - abs(x) ** exponent - abs(y) ** exponent)
        properness_checks.append({"p": p, "max_subadditivity_violation": violation})
    return {
        "C2": {"cases": finite_cases, "valid_counterexample_found": impossible_guarantees > 0, "reason": "no impossible approximation inequality found; no exhaustive-policy or lower-bound certificate"},
        "C4": {"cases": finite_cases, "valid_counterexample_found": False, "reason": "ridge augmentation remained algebraically valid; runtime subroutine is still unverified"},
        "C5": {"cases": len(properness_checks), "valid_counterexample_found": False, "reason": "no proper-loss violation found on the exhaustive scalar grid"},
        "C6": {"cases": len(properness_checks), "valid_counterexample_found": any(row["max_subadditivity_violation"] > 1e-12 for row in properness_checks), "properness_checks": properness_checks, "reason": "no counterexample found; p<=1 remains unsupported by the cited solver's stated application range"},
    }


def main() -> None:
    started = time.perf_counter()
    route1 = source_routes()
    route2 = finite_simulations()
    route3 = {
        "C2": linear_sweep(),
        "C4": linear_sweep(ridge_lambda=0.5),
        "C5": loss_sweep("huber"),
        "C6": loss_sweep("lp", p=0.5, control="single-row-support"),
    }
    route4 = falsification_routes()
    claims = {}
    for claim in ("C2", "C4", "C5", "C6"):
        claims[claim] = {
            "status": "BLOCKED",
            "confidence": "LOW",
            "routes_completed": 4,
            "route_1_proof_chain": route1[claim],
            "route_2_finite_simulation": (
                {"linear": route2["linear"], "scope": route2["scope"]} if claim == "C2"
                else {"ridge": route2["ridge"], "scope": route2["scope"]} if claim == "C4"
                else {claim: route2["huber" if claim == "C5" else "lp_p_0.5"], "scope": route2["scope"]}
            ),
            "route_3_calibrated_first_hit": route3[claim],
            "route_4_falsification": route4[claim],
            "blocker": "No executable implementation or machine-checkable proof certificate for the named quantum subroutines; finite sampling simulations cannot establish universal asymptotic runtime.",
        }
    result = {
        "paper": "arXiv:2509.24757v1",
        "source_sha256": SOURCE_SHA,
        "claims": claims,
        "runtime": {
            "seconds": time.perf_counter() - started,
            "nominal_cpu_allocation_vcpus": 8,
            "container_visible_logical_cpus": cpu_count(),
            "platform": platform.platform(),
            "python": platform.python_version(),
            "estimated_cores_before_run": 8,
            "selected_flavor": "cpu-upgrade",
        },
    }
    root = Path(__file__).resolve().parents[2]
    output = root / "outputs" / "remaining_claim_routes.json"
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
    print("REMAINING_CLAIMS_RAW_JSON=" + json.dumps(result, separators=(",", ":"), sort_keys=True))


if __name__ == "__main__":
    main()