repro-finding-most-influential-sets / code /correlated_mixture_scope_audit.py
ProCreations's picture
Expand influential-set path correlations
50d7531
Raw
History Blame
5.56 kB
#!/usr/bin/env python3
"""CPU-only correlated non-Gaussian path audit for Claims 1--3.
The earlier audit used independent product mixtures. This producer uses
non-separable two-dimensional Gaussian mixtures whose component covariances
have correlations +/-0.7. Endpoint densities are closed-form mixtures and
the path energy is a closed-form second-moment integral; only deterministic
fixed-grid quadrature is used for KL and chi-squared integrals.
"""
from __future__ import annotations
import hashlib
import json
import math
from pathlib import Path
import numpy as np
PAPER_SHA256 = "fe979c798cd48a5af02f6c647ecc29b2d6a937841adfa8ceedb492b1c2d81583"
FAMILIES = (
(
"rho-positive",
np.array([[-2.0, 0.0], [0.0, 2.0], [2.0, -1.0]]),
np.array([0.25, 0.50, 0.25]),
np.array([[[0.45, 0.315], [0.315, 0.45]], [[0.35, -0.245], [-0.245, 0.35]], [[0.55, 0.385], [0.385, 0.55]]]),
),
(
"rho-negative",
np.array([[-3.0, 1.0], [0.0, -2.0], [2.0, 2.0]]),
np.array([0.20, 0.30, 0.50]),
np.array([[[0.50, -0.35], [-0.35, 0.50]], [[0.40, 0.28], [0.28, 0.40]], [[0.30, -0.21], [-0.21, 0.30]]]),
),
(
"mixed-correlation",
np.array([[-2.0, -2.0], [1.0, 0.0], [2.0, 3.0]]),
np.array([0.30, 0.40, 0.30]),
np.array([[[0.35, 0.245], [0.245, 0.35]], [[0.60, -0.42], [-0.42, 0.60]], [[0.45, 0.315], [0.315, 0.45]]]),
),
)
def paper_sha() -> str:
return hashlib.sha256((Path(__file__).resolve().parents[1] / "source" / "paper_v1.pdf").read_bytes()).hexdigest()
def density(points: np.ndarray, means: np.ndarray, weights: np.ndarray, covariances: np.ndarray, drift: float) -> np.ndarray:
result = np.zeros(len(points), dtype=float)
scale = math.exp(drift)
variance_add = (math.exp(2.0 * drift) - 1.0) / (2.0 * drift) if abs(drift) > 1e-12 else 1.0
for mean, weight, covariance in zip(means, weights, covariances):
transformed_mean = scale * mean
transformed_covariance = scale * scale * covariance + variance_add * np.eye(2)
inverse = np.linalg.inv(transformed_covariance)
determinant = np.linalg.det(transformed_covariance)
delta = points - transformed_mean
result += weight * np.exp(-0.5 * np.einsum("ni,ij,nj->n", delta, inverse, delta)) / (2.0 * np.pi * np.sqrt(determinant))
return result
def audit() -> dict[str, object]:
if paper_sha() != PAPER_SHA256:
raise RuntimeError("paper source hash changed")
axis = np.linspace(-12.0, 12.0, 801)
grid_x, grid_y = np.meshgrid(axis, axis, indexing="ij")
points = np.stack([grid_x.ravel(), grid_y.ravel()], axis=1)
spacing = float(axis[1] - axis[0])
rows: list[dict[str, float | str]] = []
for family, means, weights, covariances in FAMILIES:
reference = np.zeros(len(points), dtype=float)
for mean, weight, covariance in zip(means, weights, covariances):
inverse = np.linalg.inv(covariance + np.eye(2))
determinant = np.linalg.det(covariance + np.eye(2))
delta = points - mean
reference += weight * np.exp(-0.5 * np.einsum("ni,ij,nj->n", delta, inverse, delta)) / (2.0 * np.pi * np.sqrt(determinant))
for drift in (-0.15, -0.10, -0.05, 0.05, 0.10, 0.15):
perturbed = density(points, means, weights, covariances, drift)
times = np.linspace(0.0, 1.0, 2001)
scale = np.exp(drift * times)
variance_add = (np.exp(2.0 * drift * times) - 1.0) / (2.0 * drift)
second_moment = []
for scalar, add in zip(scale, variance_add):
second_moment.append(sum(weight * (np.trace(scalar * scalar * covariance + add * np.eye(2)) + np.dot(scalar * mean, scalar * mean)) for mean, weight, covariance in zip(means, weights, covariances)))
energy = float(np.trapezoid(drift * drift * np.asarray(second_moment), times))
safe_reference = np.maximum(reference, np.finfo(float).tiny)
safe_perturbed = np.maximum(perturbed, np.finfo(float).tiny)
mass_reference = float(np.sum(reference) * spacing * spacing)
mass_perturbed = float(np.sum(perturbed) * spacing * spacing)
kl = float(np.sum(perturbed * np.log(safe_perturbed / safe_reference)) * spacing * spacing)
chi2 = float(np.sum(perturbed * perturbed / safe_reference) * spacing * spacing - 1.0)
rows.append({"family": family, "drift": drift, "endpoint_kl": kl, "endpoint_chi2": chi2, "path_energy": energy, "kl_over_half_energy": kl / (0.5 * energy), "chi2_over_energy": chi2 / energy, "mass_error": max(abs(mass_reference - 1.0), abs(mass_perturbed - 1.0))})
assert len(rows) == 18
assert max(row["kl_over_half_energy"] for row in rows) < 1.0
assert min(row["endpoint_chi2"] for row in rows) > 0.0
return {
"schema": "correlated-nonseparable-mixture-path-v1",
"cpu_only": True,
"paper_sha256": PAPER_SHA256,
"cells": len(rows),
"families": [family[0] for family in FAMILIES],
"drifts": [-0.15, -0.10, -0.05, 0.05, 0.10, 0.15],
"max_kl_over_half_energy": max(row["kl_over_half_energy"] for row in rows),
"min_chi2_over_energy": min(row["chi2_over_energy"] for row in rows),
"max_chi2_over_energy": max(row["chi2_over_energy"] for row in rows),
"max_mass_error": max(row["mass_error"] for row in rows),
"rows": rows,
}
if __name__ == "__main__":
print(json.dumps(audit(), indent=2, sort_keys=True))