File size: 5,327 Bytes
e6930c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | #!/usr/bin/env python3
"""Exact CPU audit of time-inhomogeneous Gaussian path perturbations.
The original claims used identity-covariance endpoint shifts. This audit
keeps the path calculation closed form but broadens the family: deterministic
time-varying drifts and linear state-dependent drifts are evaluated on several
time grids and independent dimensions. All recurrences are Fraction-valued;
only the final logarithms/exponentials are converted to floats.
"""
from __future__ import annotations
import argparse
import json
import math
from fractions import Fraction
from pathlib import Path
def deterministic_rows(dim: int, steps: int) -> list[dict[str, object]]:
dt = Fraction(1, steps)
profiles = {
"constant": [Fraction(1, 10)] * steps,
"alternating": [Fraction(1, 10) if i % 2 == 0 else Fraction(-1, 10) for i in range(steps)],
"ramp": [Fraction(2 * i - steps + 1, 10 * steps) for i in range(steps)],
"canceling": [Fraction(1, 5) if i < steps // 2 else Fraction(-1, 5) for i in range(steps)],
}
rows = []
for profile, drift in profiles.items():
mean = sum(drift, Fraction(0)) * dt
energy = Fraction(dim) * sum((u * u for u in drift), Fraction(0)) * dt
endpoint_kl = Fraction(dim) * mean * mean / 2
rows.append(
{
"family": "deterministic_drift",
"profile": profile,
"dimension": dim,
"steps": steps,
"energy": float(energy),
"endpoint_kl": float(endpoint_kl),
"kl_over_half_energy": float(endpoint_kl / (energy / 2)) if energy else 0.0,
"girsanov_upper_pass": endpoint_kl <= energy / 2,
}
)
return rows
def linear_rows(dim: int, steps: int) -> list[dict[str, object]]:
dt = Fraction(1, steps)
profiles = {
"constant_linear": [Fraction(1, 10)] * steps,
"alternating_linear": [Fraction(1, 10) if i % 2 == 0 else Fraction(-1, 10) for i in range(steps)],
"ramped_linear": [Fraction(2 * i - steps + 1, 10 * steps) for i in range(steps)],
}
rows = []
for profile, coefficients in profiles.items():
variance = Fraction(0)
path_energy = Fraction(0)
for coefficient in coefficients:
path_energy += coefficient * coefficient * variance * dt
variance = (1 + coefficient * dt) ** 2 * variance + dt
variance *= dim
q_variance = Fraction(dim)
# The scalar recurrence is independent per coordinate, so the exact
# energy and KL scale linearly with dimension.
exact_energy = path_energy * dim
endpoint_kl = (variance / q_variance - 1 - math.log(float(variance / q_variance))) * dim / 2
variance_ratio = float(variance / q_variance)
chi_square = (1.0 / math.sqrt(variance_ratio * (2.0 - variance_ratio))) ** dim - 1.0
rows.append(
{
"family": "linear_state_drift",
"profile": profile,
"dimension": dim,
"steps": steps,
"terminal_variance_ratio": variance_ratio,
"energy": float(exact_energy),
"endpoint_kl": float(endpoint_kl),
"chi_square": chi_square,
"chi2_over_energy": chi_square / float(exact_energy) if exact_energy else 0.0,
"girsanov_upper_pass": endpoint_kl <= exact_energy / 2 + 1e-15,
"chi_square_finite": variance_ratio < 2.0,
}
)
return rows
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--output-dir", type=Path, required=True)
args = parser.parse_args()
args.output_dir.mkdir(parents=True, exist_ok=True)
rows = []
for dim in (1, 2, 8, 32):
for steps in (4, 8, 16, 32, 64, 128):
rows.extend(deterministic_rows(dim, steps))
rows.extend(linear_rows(dim, steps))
linear = [row for row in rows if row["family"] == "linear_state_drift"]
summary = {
"cells": len(rows),
"dimensions": [1, 2, 8, 32],
"steps": [4, 8, 16, 32, 64, 128],
"families": ["deterministic_drift", "linear_state_drift"],
"max_kl_over_half_energy": max(float(row["kl_over_half_energy"]) for row in rows if "kl_over_half_energy" in row),
"max_linear_endpoint_kl_over_half_energy": max(
2 * float(row["endpoint_kl"]) / float(row["energy"]) for row in linear if row["energy"]
),
"chi2_over_energy_range": [min(float(row["chi2_over_energy"]) for row in linear), max(float(row["chi2_over_energy"]) for row in linear)],
"all_girsanov_upper_pass": all(bool(row["girsanov_upper_pass"]) for row in rows),
"all_linear_chi_square_finite": all(bool(row["chi_square_finite"]) for row in linear),
"exact_fraction_recurrence": True,
}
(args.output_dir / "path_gaussian_scope.json").write_text(
json.dumps({"summary": summary, "rows": rows}, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
print(json.dumps(summary, indent=2, sort_keys=True))
return 0 if summary["all_girsanov_upper_pass"] and summary["all_linear_chi_square_finite"] else 2
if __name__ == "__main__":
raise SystemExit(main())
|