Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Clean-room checks for the three spectral knowledge-transfer claims.""" | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| from pathlib import Path | |
| import numpy as np | |
| def schedule(total: int, gamma: float) -> np.ndarray: | |
| phase = max(1, int(total / np.log2(total))) | |
| return gamma * 2.0 ** -(np.arange(total) // phase) | |
| def projectors_identity(cells: int, seed: int = 19) -> dict: | |
| """Appendix geometric-consistency identity, using noncommuting subspaces.""" | |
| rng = np.random.default_rng(seed) | |
| largest = 0.0 | |
| for _ in range(cells): | |
| dimension, teacher_rank, student_rank = 12, 8, 6 | |
| teacher, _ = np.linalg.qr(rng.normal(size=(dimension, teacher_rank))) | |
| student, _ = np.linalg.qr(rng.normal(size=(dimension, student_rank))) | |
| pt, ps = teacher @ teacher.T, student @ student.T | |
| truth = rng.normal(size=dimension) | |
| direct = 0.5 * np.linalg.norm(ps @ truth - truth) ** 2 | |
| transfer = 0.5 * np.linalg.norm(ps @ pt @ truth - truth) ** 2 | |
| expected = 0.5 * np.linalg.norm(ps @ (np.eye(dimension) - pt) @ truth) ** 2 | |
| largest = max(largest, abs((transfer - direct) - expected)) | |
| return {"noncommuting_subspace_cells": cells, "max_static_gap_identity_error": largest} | |
| def rate_audits() -> dict: | |
| """Independent numerical audit of the exponents in Theorems 4.2 and 5.2.""" | |
| alpha_t = np.linspace(1.05, 3.0, 200) | |
| alpha_s = np.linspace(1.1, 4.0, 200) | |
| beta = np.linspace(0.0, 0.9, 10) | |
| at, ass, be = np.meshgrid(alpha_t, alpha_s, beta, indexing="ij") | |
| valid = (at > 1 + be) & (ass > at) | |
| kappa = (at - 1 - be) * (1 / at - 1 / ass) | |
| horizon_order = 10_000.0 ** (1 / at) > 10_000.0 ** (1 / ass) | |
| kd_violations = int(np.sum(valid & ((kappa <= 0) | ~horizon_order))) | |
| delta = 2 * ass / (at * (2 * ass + 1)) | |
| w2s_valid = (at > 1) & (ass > 1) | |
| w2s_violations = int(np.sum(w2s_valid & ((delta <= 0) | (delta >= 1)))) | |
| return {"kd_admissible_cells": int(valid.sum()), "kd_positive_exponent_violations": kd_violations, | |
| "w2s_admissible_cells": int(w2s_valid.sum()), "w2s_positive_rate_violations": w2s_violations} | |
| def run_sgd(*, alpha_teacher: float, alpha_student: float, original: int, transfer: int, | |
| trials: int, noise: float, seed: int, signal_dimensions: int | None = None) -> dict: | |
| """Paper protocol: fresh single samples, teacher then direct/distilled student.""" | |
| rng = np.random.default_rng(seed) | |
| d = 100 | |
| modes = np.arange(1, d + 1, dtype=float) | |
| lt = modes ** -alpha_teacher | |
| ls = modes ** -alpha_student | |
| st, ss = np.sqrt(lt), np.sqrt(ls) | |
| teacher = np.zeros((trials, d)); direct = np.zeros((trials, d)); distilled = np.zeros((trials, d)) | |
| truth = rng.normal(size=(trials, d)) | |
| if signal_dimensions is not None: | |
| truth[:, signal_dimensions:] = 0.0 | |
| # The signal is in the teacher feature coordinates, exactly as in the source. | |
| for lr in schedule(original, 0.01): | |
| x = rng.normal(size=(trials, d)); phi_t, phi_s = x * st, x * ss | |
| y = np.sum(phi_t * truth, axis=1) + noise * rng.normal(size=trials) | |
| teacher += lr * (y - np.sum(phi_t * teacher, axis=1))[:, None] * phi_t | |
| direct += lr * (y - np.sum(phi_s * direct, axis=1))[:, None] * phi_s | |
| for lr in schedule(transfer, 0.01): | |
| x = rng.normal(size=(trials, d)); phi_t, phi_s = x * st, x * ss | |
| pseudo = np.sum(phi_t * teacher, axis=1) | |
| distilled += lr * (pseudo - np.sum(phi_s * distilled, axis=1))[:, None] * phi_s | |
| # Exact population risks in the common latent Gaussian basis. | |
| teacher_risk = np.sum((st * (teacher - truth)) ** 2, axis=1) | |
| direct_risk = np.sum((ss * direct - st * truth) ** 2, axis=1) | |
| distilled_risk = np.sum((ss * distilled - st * truth) ** 2, axis=1) | |
| return {"teacher_risk": teacher_risk, "direct_risk": direct_risk, "distilled_risk": distilled_risk} | |
| def w2s_early_stop_scan(trials: int, seed: int, noise: float = 1.0) -> dict: | |
| """Mechanism audit: save student checkpoints before tail-noise memorization.""" | |
| rng = np.random.default_rng(seed) | |
| d, original, maximum = 100, 2000, 100_000 | |
| modes = np.arange(1, d + 1, dtype=float) | |
| st, ss = modes ** (-1.5 / 2), modes ** (-1.0 / 2) | |
| truth = rng.normal(size=(trials, d)); truth[:, 10:] = 0.0 | |
| teacher = np.zeros((trials, d)); student = np.zeros((trials, d)) | |
| for lr in schedule(original, .01): | |
| x = rng.normal(size=(trials, d)); pt = x * st | |
| y = np.sum(pt * truth, axis=1) + noise * rng.normal(size=trials) | |
| teacher += lr * (y - np.sum(pt * teacher, axis=1))[:, None] * pt | |
| checkpoints = {100, 500, 2_000, 10_000, maximum} | |
| values = {} | |
| for step, lr in enumerate(schedule(maximum, .01), 1): | |
| x = rng.normal(size=(trials, d)); pt, ps = x * st, x * ss | |
| pseudo = np.sum(pt * teacher, axis=1) | |
| student += lr * (pseudo - np.sum(ps * student, axis=1))[:, None] * ps | |
| if step in checkpoints: | |
| values[str(step)] = np.sum((ss * student - st * truth) ** 2, axis=1) | |
| teacher_risk = np.sum((st * (teacher - truth)) ** 2, axis=1) | |
| matrix = np.stack([values[str(k)] for k in sorted(checkpoints)], axis=1) | |
| best = matrix.min(axis=1) | |
| return {"checkpoints": sorted(checkpoints), "median_risks": {str(k): float(np.median(values[str(k)])) for k in sorted(checkpoints)}, | |
| "mean_risks": {str(k): float(np.mean(values[str(k)])) for k in sorted(checkpoints)}, | |
| "best_student_beats_teacher_fraction": float(np.mean(best < teacher_risk)), | |
| "median_teacher_risk": float(np.median(teacher_risk)), "mean_teacher_risk": float(np.mean(teacher_risk)), | |
| "median_best_student_risk": float(np.median(best)), "mean_best_student_risk": float(np.mean(best))} | |
| def main() -> None: | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--output", default="outputs/independent_verification.json") | |
| parser.add_argument("--trials", type=int, default=64) | |
| parser.add_argument("--transfer", type=int, default=500000) | |
| args = parser.parse_args() | |
| geometric, rates = projectors_identity(9000), rate_audits() | |
| kd = run_sgd(alpha_teacher=1.0, alpha_student=1.5, original=2000, transfer=args.transfer, | |
| trials=args.trials, noise=.1, seed=260601292) | |
| der = kd["direct_risk"] / kd["distilled_risk"] | |
| # W2S control: low-dimensional target is created by retaining only its head. | |
| w2s = w2s_early_stop_scan(args.trials, 260601293) | |
| result = {"paper": "ykWN4LG9vE", "protocol": {"d": 100, "N": 2000, "n": args.transfer, | |
| "trials": args.trials, "alpha_teacher": 1.0, "alpha_student": 1.5, "noise_std": .1}, | |
| "claim_1_geometric_decomposition": geometric, "claim_2_kd_horizon": {**rates, | |
| "median_teacher_risk": float(np.median(kd['teacher_risk'])), "median_direct_risk": float(np.median(kd['direct_risk'])), | |
| "median_distilled_risk": float(np.median(kd['distilled_risk'])), "minimum_der": float(der.min()), "median_der": float(np.median(der))}, | |
| "claim_3_w2s_denoising": {"low_intrinsic_dimension": 10, **w2s}} | |
| target = Path(args.output); target.parent.mkdir(parents=True, exist_ok=True) | |
| target.write_text(json.dumps(result, indent=2) + "\n") | |
| print(json.dumps(result, indent=2)) | |
| if __name__ == '__main__': | |
| main() | |