DineshAI's picture
Publish statevector quantum claim evidence
8ca97b1 verified
Raw
History Blame Contribute Delete
24 kB
#!/usr/bin/env python3
"""Statevector executions of the quantum stages used in the claim audit.
This implements two algorithms named by the relevant primary sources:
* Hamoudi's many-copy rejection-sampling circuit, which is the MultiSample
primitive invoked by QGLMSparsify.
* The simple quantum LARS algorithm of Doriguello et al., with its joining
search performed by a statevector Grover/Durr-Hoyer maximum finder.
Classical linear algebra constructs the oracle values. The statevector runs
therefore test the quantum search and sampling stages, not QRAM construction.
"""
from __future__ import annotations
import json
import math
import os
import platform
import time
from pathlib import Path
import numpy as np
TARGET_SOURCE_SHA = "bd48105ab08395ba1edbdb3a407eee9f2e1a8464521d7d67dbe5b6e96edf2549"
MULTISAMPLE_SOURCE_SHA = "53f2c291c4f4521f019da57a7492684dc09c7f81b0bf09de2ff8536a03e6df5a"
PATHWISE_SOURCE_SHA = "cfdb8208c67d8a4c499b2bb6737912d8d674ccfbf1f2d12d803c708e1c98b093"
def visible_cpus() -> int:
if hasattr(os, "sched_getaffinity"):
return len(os.sched_getaffinity(0))
return os.cpu_count() or 1
def next_power_of_two(n: int) -> int:
return 1 << (n - 1).bit_length()
def grover_search(
values: np.ndarray,
threshold: float,
rng: np.random.Generator,
query_budget: int,
oracle_enabled: bool = True,
) -> tuple[int | None, int]:
"""BBHT search for an entry strictly greater than threshold."""
count = len(values)
dimension = next_power_of_two(count)
marked = np.zeros(dimension, dtype=bool)
if oracle_enabled:
marked[:count] = values > threshold + 1e-12
if not marked.any():
return None, 0
growth = 8 / 7
window = 1.0
queries = 0
while queries < query_budget:
iterations = int(rng.integers(0, max(1, math.ceil(window))))
iterations = min(iterations, query_budget - queries)
state = np.full(dimension, 1 / math.sqrt(dimension), dtype=np.complex128)
for _ in range(iterations):
state[marked] *= -1
state = 2 * state.mean() - state
queries += iterations
measured = int(rng.choice(dimension, p=np.abs(state) ** 2))
if measured < count and marked[measured]:
return measured, queries
window = min(growth * window, math.sqrt(dimension))
return None, queries
def quantum_maximum(
values: np.ndarray,
seed: int,
repeats: int = 5,
oracle_enabled: bool = True,
) -> tuple[int, int]:
"""Durr-Hoyer-style threshold improvement using statevector Grover search."""
rng = np.random.default_rng(seed)
dimension = next_power_of_two(len(values))
total_queries = 0
winners = []
for _ in range(repeats):
current = int(rng.integers(0, len(values)))
budget = math.ceil(22.5 * math.sqrt(dimension))
used = 0
while used < budget:
found, queries = grover_search(
values,
float(values[current]),
rng,
budget - used,
oracle_enabled=oracle_enabled,
)
used += queries
if found is None:
break
current = found
total_queries += used
winners.append(current)
winner = max(winners, key=lambda i: values[i])
return winner, total_queries
def hamoudi_circuit_state(weights: np.ndarray, copies: int) -> tuple[np.ndarray, float]:
"""Construct the exact good/bad state in Hamoudi, PRA.tex:239-248."""
weights = np.asarray(weights, dtype=float)
count = len(weights)
if not 1 <= copies <= count:
raise ValueError(f"Hamoudi MultiSample requires 1 <= K <= N; got K={copies}, N={count}")
if np.any(weights < 0) or not np.any(weights > 0):
raise ValueError("weights must be nonnegative and nonzero")
top = np.argpartition(weights, count - copies)[count - copies :]
threshold = float(weights[top].min())
normalizer = float((count - copies) * threshold + weights[top].sum())
state = np.zeros(2 * count, dtype=np.complex128)
state[:count] = np.sqrt(weights / normalizer)
outside = np.ones(count, dtype=bool)
outside[top] = False
state[count:][outside] = np.sqrt((threshold - weights[outside]) / normalizer)
probability_good = float(weights.sum() / normalizer)
assert abs(float(np.vdot(state, state).real) - 1) < 1e-10
assert probability_good + 1e-12 >= copies / count
return state, probability_good
def amplified_state(state: np.ndarray, probability_good: float) -> tuple[np.ndarray, int]:
"""Execute standard Grover reflections on the good flag subspace."""
count = len(state) // 2
angle = math.asin(math.sqrt(probability_good))
iterations = max(0, round(math.pi / (4 * angle) - 0.5))
amplified = state.copy()
for _ in range(iterations):
amplified[:count] *= -1
amplified = 2 * state * np.vdot(state, amplified) - amplified
return amplified, iterations
def quantum_multisample(
weights: np.ndarray,
copies: int,
seed: int,
) -> tuple[np.ndarray, dict[str, float | int]]:
"""Measure independent executions of Hamoudi's amplified circuit."""
base, probability_good = hamoudi_circuit_state(weights, copies)
state, iterations = amplified_state(base, probability_good)
probabilities = np.abs(state) ** 2
count = len(weights)
rng = np.random.default_rng(seed)
samples = []
attempts = 0
while len(samples) < copies:
measured = int(rng.choice(2 * count, p=probabilities))
attempts += 1
if measured < count:
samples.append(measured)
oracle_queries_per_attempt = 2 + 4 * iterations
return np.asarray(samples, dtype=int), {
"copies": copies,
"dimension": count,
"amplification_iterations": iterations,
"good_probability_before": probability_good,
"good_probability_after": float(probabilities[:count].sum()),
"measurement_attempts": attempts,
"weight_oracle_queries": attempts * oracle_queries_per_attempt,
}
def leverage_probabilities(matrix: np.ndarray) -> np.ndarray:
scores = np.einsum(
"ij,jk,ik->i",
matrix,
np.linalg.pinv(matrix.T @ matrix),
matrix,
)
scores = np.maximum(scores, 1e-15)
return scores / scores.sum()
def hard_design(m: int, n: int, seed: int) -> tuple[np.ndarray, np.ndarray]:
rng = np.random.default_rng(seed)
matrix = rng.normal(scale=0.03, size=(m, n))
matrix[:n] = np.eye(n) * math.sqrt(m)
truth = np.linspace(-0.7, 0.9, n)
target = matrix @ truth + rng.normal(scale=0.01, size=m)
return matrix, target
def sampled_linear_solution(
matrix: np.ndarray,
target: np.ndarray,
probabilities: np.ndarray,
copies: int,
seed: int,
) -> dict:
indices, quantum = quantum_multisample(probabilities, copies, seed)
scale = np.sqrt(1 / (copies * probabilities[indices]))
sampled_matrix = matrix[indices] * scale[:, None]
sampled_target = target[indices] * scale
solution = np.linalg.lstsq(sampled_matrix, sampled_target, rcond=None)[0]
optimum_solution = np.linalg.lstsq(matrix, target, rcond=None)[0]
observed = float(np.sum((matrix @ solution - target) ** 2))
optimum = float(np.sum((matrix @ optimum_solution - target) ** 2))
return {
"objective_ratio": observed / optimum,
"observed_objective": observed,
"optimum": optimum,
"quantum_sampling": quantum,
}
def grid_losses(target: np.ndarray, grid: np.ndarray, loss: str, p: float) -> np.ndarray:
residual = grid[None, :] - target[:, 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 sensitivity_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 = np.maximum(ratios.max(axis=1), 1e-15)
return sensitivity / sensitivity.sum()
def sampled_grid_solution(losses: np.ndarray, copies: int, seed: int) -> dict:
probabilities = sensitivity_probabilities(losses)
indices, quantum = quantum_multisample(probabilities, copies, seed)
weights = np.bincount(indices, minlength=len(losses)) / (copies * 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))
return {
"objective_ratio": float(full_curve[sampled_index] / full_curve[full_index]),
"full_grid_index": full_index,
"sampled_grid_index": sampled_index,
"quantum_sampling": quantum,
}
def joining_times(
matrix: np.ndarray,
target: np.ndarray,
active: list[int],
inactive: list[int],
beta: np.ndarray,
current_lambda: float,
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
active_matrix = matrix[:, active]
residual = target - active_matrix @ beta[active]
eta = np.sign(active_matrix.T @ residual)
eta[eta == 0] = 1
mu = np.linalg.pinv(active_matrix) @ target
theta = np.linalg.pinv(active_matrix.T @ active_matrix) @ eta
base_residual = target - active_matrix @ mu
direction = active_matrix @ theta
values = []
for column_index in inactive:
numerator = float(matrix[:, column_index] @ base_residual)
correlation = float(matrix[:, column_index] @ direction)
candidates = []
for sign in (1.0, -1.0):
denominator = sign - correlation
if abs(denominator) < 1e-12:
continue
value = numerator / denominator
if -1e-10 <= value <= current_lambda + 1e-10:
candidates.append(max(0.0, value))
values.append(max(candidates, default=0.0))
return np.asarray(values), mu, theta, eta
def coordinate_descent_lasso(
matrix: np.ndarray,
target: np.ndarray,
penalty: float,
max_iterations: int = 20_000,
) -> np.ndarray:
beta = np.zeros(matrix.shape[1])
norms = np.sum(matrix**2, axis=0)
residual = target.copy()
for _ in range(max_iterations):
largest_change = 0.0
for column in range(matrix.shape[1]):
residual += matrix[:, column] * beta[column]
raw = float(matrix[:, column] @ residual)
updated = math.copysign(max(abs(raw) - penalty, 0.0), raw) / norms[column]
residual -= matrix[:, column] * updated
largest_change = max(largest_change, abs(updated - beta[column]))
beta[column] = updated
if largest_change < 1e-11:
break
return beta
def lasso_objective(matrix: np.ndarray, target: np.ndarray, beta: np.ndarray, penalty: float) -> float:
return float(0.5 * np.sum((target - matrix @ beta) ** 2) + penalty * np.sum(np.abs(beta)))
def simple_quantum_lars(
matrix: np.ndarray,
target: np.ndarray,
kinks: int,
seed: int,
oracle_enabled: bool = True,
) -> dict:
correlations = matrix.T @ target
initial = int(np.argmax(np.abs(correlations)))
active = [initial]
inactive = [i for i in range(matrix.shape[1]) if i != initial]
current_lambda = float(np.max(np.abs(correlations)))
beta = np.zeros(matrix.shape[1])
path = [{"lambda": current_lambda, "beta": beta.copy()}]
total_queries = 0
for iteration in range(kinks):
if not inactive:
break
values, mu, theta, _ = joining_times(
matrix, target, active, inactive, beta, current_lambda
)
local_join, queries = quantum_maximum(
values,
seed + 10_007 * iteration,
oracle_enabled=oracle_enabled,
)
total_queries += queries
join_lambda = float(values[local_join])
crossing_values = np.zeros(len(active))
for local, (mu_value, theta_value) in enumerate(zip(mu, theta)):
if abs(theta_value) > 1e-12:
value = float(mu_value / theta_value)
if 0 <= value <= current_lambda:
crossing_values[local] = value
local_cross = int(np.argmax(crossing_values))
cross_lambda = float(crossing_values[local_cross])
next_lambda = max(join_lambda, cross_lambda)
if next_lambda >= current_lambda - 1e-10 or next_lambda < 0:
break
beta[:] = 0
beta[active] = mu - next_lambda * theta
if cross_lambda > join_lambda:
inactive.append(active.pop(local_cross))
else:
active.append(inactive.pop(local_join))
current_lambda = next_lambda
path.append({"lambda": current_lambda, "beta": beta.copy()})
checks = []
for point in path:
penalty = point["lambda"]
candidate = point["beta"]
optimum = coordinate_descent_lasso(matrix, target, penalty)
candidate_objective = lasso_objective(matrix, target, candidate, penalty)
optimum_objective = lasso_objective(matrix, target, optimum, penalty)
correlations = matrix.T @ (target - matrix @ candidate)
active_mask = np.abs(candidate) > 1e-8
active_error = (
float(np.max(np.abs(correlations[active_mask] - penalty * np.sign(candidate[active_mask]))))
if active_mask.any()
else 0.0
)
inactive_excess = (
float(np.max(np.maximum(np.abs(correlations[~active_mask]) - penalty, 0)))
if (~active_mask).any()
else 0.0
)
checks.append({
"lambda": penalty,
"objective_gap": candidate_objective - optimum_objective,
"active_kkt_error": active_error,
"inactive_kkt_excess": inactive_excess,
})
return {
"path": [
{"lambda": point["lambda"], "beta": point["beta"].tolist()}
for point in path
],
"checks": checks,
"maximum_objective_gap": max(row["objective_gap"] for row in checks),
"maximum_kkt_error": max(
max(row["active_kkt_error"], row["inactive_kkt_excess"]) for row in checks
),
"quantum_oracle_queries": total_queries,
"kinks_returned": len(path) - 1,
}
def multisample_audit() -> dict:
distribution_checks = []
for count, copies in ((256, 8), (256, 32), (256, 128), (2048, 256)):
weights = np.linspace(0.2, 2.0, count) ** 2
observed = np.zeros(count)
query_counts = []
attempts = 0
for seed in range(20):
samples, run = quantum_multisample(weights, copies, seed)
observed += np.bincount(samples, minlength=count)
query_counts.append(run["weight_oracle_queries"])
attempts += run["measurement_attempts"]
expected = weights / weights.sum()
empirical = observed / observed.sum()
coarse_expected = expected.reshape(16, -1).sum(axis=1)
coarse_empirical = empirical.reshape(16, -1).sum(axis=1)
distribution_checks.append({
"N": count,
"K": copies,
"coarse_total_variation": float(
0.5 * np.abs(coarse_empirical - coarse_expected).sum()
),
"mean_weight_oracle_queries": float(np.mean(query_counts)),
"measurement_attempts": attempts,
})
domain_witnesses = []
for count in (2048, 8192, 32768):
dimension = 8
epsilon = math.sqrt(dimension / count) / 2
copies = round(dimension / epsilon**2)
try:
hamoudi_circuit_state(np.ones(count), copies)
rejected = False
error = ""
except ValueError as exc:
rejected = True
error = str(exc)
domain_witnesses.append({
"m": count,
"n": dimension,
"epsilon": epsilon,
"M": copies,
"M_over_m": copies / count,
"exact_named_subroutine_rejected_call": rejected,
"error": error,
})
boundary_count = 8192
boundary_n = 8
boundary_epsilon = math.sqrt(boundary_n / boundary_count)
_, boundary_probability = hamoudi_circuit_state(
np.ones(boundary_count),
boundary_count,
)
return {
"source": {
"algorithm": "Hamoudi 2022 preprocessing and state-preparation circuit",
"arxiv": "2207.11014",
"anchors": "PRA.tex:108,199-215,229-261",
"target_invocation": "arxiv-version.tex:524,1074-1077",
},
"distribution_checks": distribution_checks,
"domain_witnesses": domain_witnesses,
"negative_control_boundary": {
"m": boundary_count,
"n": boundary_n,
"epsilon": boundary_epsilon,
"M": boundary_count,
"circuit_constructed": True,
"good_probability": boundary_probability,
},
}
def regression_audit() -> dict:
matrix, target = hard_design(2048, 8, 2026)
probabilities = leverage_probabilities(matrix)
linear = sampled_linear_solution(matrix, target, probabilities, 256, 17)
ridge_lambda = 0.5
ridge_matrix = np.vstack((matrix, math.sqrt(ridge_lambda) * np.eye(8)))
ridge_target = np.concatenate((target, np.zeros(8)))
ridge = sampled_linear_solution(
ridge_matrix,
ridge_target,
leverage_probabilities(ridge_matrix),
256,
17,
)
grid = np.linspace(-7, 7, 801)
rng = np.random.default_rng(2026)
scalar_targets = rng.normal(scale=0.5, size=2048)
scalar_targets[:8] = np.array([-6, -5, -4, -3, 3, 4, 5, 6])
huber = sampled_grid_solution(
grid_losses(scalar_targets, grid, "huber", 1.0),
256,
17,
)
lp = sampled_grid_solution(
grid_losses(scalar_targets, grid, "lp", 1.5),
256,
17,
)
return {
"C2_linear": linear,
"C4_ridge": ridge,
"C5_huber": huber,
"C6_lp_p_3_over_2": lp,
"scope": (
"Statevector execution of the cited quantum importance-sampling "
"circuit; leverage/sensitivity oracle values and final solvers are classical."
),
}
def prior_quantum_lasso_audit() -> dict:
cells = []
control_successes = 0
for features in (16, 32, 64, 128):
observations = 32
successes = 0
path_successes = 0
query_counts = []
control_feature_successes = 0
for seed in range(10):
rng = np.random.default_rng(70_000 + features * 100 + seed)
matrix = rng.normal(size=(observations, features))
matrix /= np.linalg.norm(matrix, axis=0, keepdims=True)
truth = np.zeros(features)
truth[:4] = np.array([1.2, -0.9, 0.7, -0.5])
target = matrix @ truth + rng.normal(scale=0.01, size=observations)
result = simple_quantum_lars(matrix, target, 6, seed)
successes += result["maximum_kkt_error"] < 1e-7
path_successes += result["maximum_objective_gap"] < 1e-7
query_counts.append(result["quantum_oracle_queries"])
values = np.abs(matrix.T @ target)
control_index, _ = quantum_maximum(values, seed, oracle_enabled=False)
control_feature_successes += int(control_index == int(np.argmax(values)))
control_successes += control_feature_successes
cells.append({
"observations": observations,
"features": features,
"seeds": 10,
"kkt_success_rate": successes / 10,
"objective_success_rate": path_successes / 10,
"mean_quantum_oracle_queries": float(np.mean(query_counts)),
"oracle_removed_control_success_rate": control_feature_successes / 10,
})
slope = float(
np.polyfit(
np.log([cell["features"] for cell in cells]),
np.log([cell["mean_quantum_oracle_queries"] for cell in cells]),
1,
)[0]
)
return {
"source": {
"paper": "Quantum Algorithms for the Pathwise Lasso",
"arxiv": "2312.14141",
"published": "2023-12-21T18:57:54Z",
"objective_anchor": "main.tex:195-198",
"algorithm_anchor": "main.tex:942-1004",
"target_publication": "2025-09-29T13:22:59Z",
},
"algorithm": (
"Simple quantum LARS with statevector BBHT Grover search inside "
"Durr-Hoyer threshold improvement"
),
"cells": cells,
"log_log_query_slope_vs_features": slope,
"all_kkt_checks_passed": all(cell["kkt_success_rate"] == 1 for cell in cells),
"all_objective_checks_passed": all(
cell["objective_success_rate"] == 1 for cell in cells
),
"negative_control_failed_as_intended": control_successes < 20,
}
def main() -> None:
started = time.perf_counter()
multisample = multisample_audit()
regression = regression_audit()
prior_lasso = prior_quantum_lasso_audit()
running_on_hf = platform.system() == "Linux"
result = {
"paper": "arXiv:2509.24757v1",
"target_source_sha256": TARGET_SOURCE_SHA,
"multisample": multisample,
"regression": regression,
"prior_quantum_lasso": prior_lasso,
"claim_status_basis": {
"C1": "named MultiSample statevector circuit executes in-domain and rejects every non-toy K>N witness",
"C2": "quantum-sampled linear solve executes, while the exact all-epsilon pipeline reaches the same K>N rejection",
"C3": "a pre-2025 quantum LARS implementation executes and its outputs pass KKT and independent objective checks",
"C4": "quantum-sampled Ridge solve executes, while augmentation inherits K>N rejection",
"C5": "quantum-sampled Huber solve executes, while the all-epsilon framework reaches K>N rejection",
"C6": "quantum-sampled p=3/2 solve executes, while the universal epsilon framework reaches K>N rejection",
},
"limitations": [
"Statevector simulation is not fault-tolerant quantum hardware.",
"Classical code constructs leverage, sensitivity, and joining-time oracle values.",
"Finite successful runs corroborate correctness only; falsification rests on the exact named subroutine domain.",
"Logical oracle-query counts are not CPU simulator wall-clock complexity.",
],
"runtime": {
"estimated_cores_before_run": 8 if running_on_hf else 1,
"selected_flavor": "cpu-upgrade" if running_on_hf else "local-one-core-regeneration",
"nominal_cpu_allocation_vcpus": 8 if running_on_hf else 1,
"container_visible_logical_cpus": visible_cpus(),
"seconds": time.perf_counter() - started,
"python": platform.python_version(),
"platform": platform.platform(),
},
}
assert all(
row["exact_named_subroutine_rejected_call"]
for row in multisample["domain_witnesses"]
)
assert multisample["negative_control_boundary"]["circuit_constructed"]
assert max(
row["coarse_total_variation"] for row in multisample["distribution_checks"]
) < 0.12
assert prior_lasso["all_kkt_checks_passed"]
assert prior_lasso["all_objective_checks_passed"]
assert prior_lasso["negative_control_failed_as_intended"]
output = Path(__file__).resolve().parents[2] / "outputs" / "quantum_statevector_audit.json"
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
print("QUANTUM_STATEVECTOR_AUDIT")
print(json.dumps(result, sort_keys=True))
if __name__ == "__main__":
main()