Buckets:
| #!/usr/bin/env python3 | |
| """Task 7 prototype: subspace error geometry & precision-weighted Mahalanobis. | |
| Question: are per-category "failure" rankings an artifact of latent-space | |
| anisotropy? Euclidean MSE weights every direction equally; if the prediction-error | |
| cloud is anisotropic, MSE is dominated by a few high-variance directions. | |
| Algorithm: | |
| 1. clean forward over the SAME 1000-sample/50-class subset as task 4 (same seed). | |
| 2. error vectors e = prediction - target (1280-d encoder space), per target token. | |
| 3. PCA / eigendecomposition of the error covariance -> principal error directions | |
| and their variances (the anisotropy of the error space). | |
| 4. precision-weighted Mahalanobis: D_M^2(e) = e^T Sigma^-1 e = sum_k (e.v_k)^2/lambda_k | |
| -- reweights each direction by 1/variance. | |
| 5. rank the 50 categories by mean MSE vs mean D_M; Spearman rho + biggest movers. | |
| Disagreement => the failure ranking is sensitive to anisotropy. | |
| """ | |
| # ruff: noqa: N806 -- names mirror this analysis's linear-algebra notation | |
| # (e.g. E for error matrix, D for dim, V for eigenvectors) rather than PEP8 casing. | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import matplotlib | |
| import numpy as np | |
| import yaml | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| from scipy.stats import spearmanr | |
| SP = str(Path(__file__).resolve().parents[2] / "scripts") | |
| sys.path.insert(0, SP) | |
| from run_deep import HUMAN, OUT, collect_clean, load_model, masks, seed_all # noqa: E402 | |
| MANIFEST = OUT / "ijepa_task4/20260904T213720Z/dataset_manifest.json" # same subset/seed as task 4 | |
| def main(): | |
| seed_all() | |
| a = load_model() | |
| context_ids, target_ids, n_patches = masks(a) | |
| samples = json.load(open(MANIFEST)) # identical 1000 samples task 4 used | |
| N = len(samples) | |
| print( | |
| f"task7: {N} samples, {len({s['class_name'] for s in samples})} classes (same subset as task 4)", | |
| flush=True, | |
| ) | |
| clean = collect_clean(a, samples, context_ids, target_ids, []) # no per-layer caching | |
| pred = clean["pred"].float().numpy() # [N,16,1280] | |
| tgt = clean["tgt"].float().numpy() # [N,16,1280] | |
| E = pred - tgt # error vectors | |
| D = E.shape[-1] | |
| T = E.shape[1] | |
| Et = E.reshape(-1, D) # [N*16, 1280] per-token errors (full rank cov) | |
| print(f"error vectors: {Et.shape[0]} tokens x {D} dims", flush=True) | |
| # covariance of the error cloud (centered) + eigendecomposition = PCA | |
| mu = Et.mean(0) | |
| Ec = Et - mu | |
| cov = np.cov(Ec.T) # [D,D] | |
| lam, V = np.linalg.eigh(cov) # ascending | |
| lam = lam[::-1].copy() | |
| V = V[:, ::-1].copy() | |
| lam = np.clip(lam, 1e-10, None) | |
| # anisotropy metrics of the error space | |
| p = lam / lam.sum() | |
| aniso = { | |
| "participation_ratio": float((lam.sum() ** 2) / np.sum(lam**2)), | |
| "effective_rank_entropy": float(np.exp(-(p * np.log(p)).sum())), | |
| "top1_var_frac": float(lam[0] / lam.sum()), | |
| "top10_var_frac": float(lam[:10].sum() / lam.sum()), | |
| "top50_var_frac": float(lam[:50].sum() / lam.sum()), | |
| "eigval_ratio_max_min": float(lam[0] / lam[-1]), | |
| "isotropy_score_min_over_max": float(lam[-1] / lam[0]), # 1.0 == isotropic | |
| "dim": int(D), | |
| } | |
| # precision-weighted Mahalanobis of the RAW error (distance from perfect=0), | |
| # in the eigenbasis: D_M^2 = sum (e.v_k)^2 / lam_k | |
| proj = Et @ V # raw error projected (from origin) | |
| dm2 = (proj**2 / lam).sum(1) # [N*16] | |
| dm = np.sqrt(np.clip(dm2, 0, None)) | |
| mse = (Et**2).sum(1) # Euclidean squared error per token | |
| # aggregate per category (token-level -> category mean) | |
| tok_labels = np.repeat([s["class_name"] for s in samples], T) | |
| cats = sorted(set(tok_labels), key=lambda x: int(x)) | |
| cat_mse = {c: float(mse[tok_labels == c].mean()) for c in cats} | |
| cat_dm = {c: float(dm[tok_labels == c].mean()) for c in cats} | |
| # rankings (1 = worst / highest) | |
| order_mse = sorted(cats, key=lambda c: -cat_mse[c]) | |
| order_dm = sorted(cats, key=lambda c: -cat_dm[c]) | |
| rank_mse = {c: i + 1 for i, c in enumerate(order_mse)} | |
| rank_dm = {c: i + 1 for i, c in enumerate(order_dm)} | |
| rho, pval = spearmanr([rank_mse[c] for c in cats], [rank_dm[c] for c in cats]) | |
| movers = sorted(cats, key=lambda c: -abs(rank_mse[c] - rank_dm[c])) | |
| ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") | |
| rd = OUT / "ijepa_task7" / ts | |
| rd.mkdir(parents=True, exist_ok=True) | |
| result = { | |
| "n_samples": N, | |
| "n_error_vectors": int(Et.shape[0]), | |
| "dim": int(D), | |
| "anisotropy": aniso, | |
| "spearman_mse_vs_maha": {"rho": float(rho), "pval": float(pval)}, | |
| "eigenvalues_top50": lam[:50].tolist(), | |
| "category_mse": cat_mse, | |
| "category_maha": cat_dm, | |
| "rank_mse": rank_mse, | |
| "rank_maha": rank_dm, | |
| "biggest_rank_movers": [ | |
| { | |
| "class": c, | |
| "human": HUMAN.get(c, ""), | |
| "rank_mse": rank_mse[c], | |
| "rank_maha": rank_dm[c], | |
| "shift": rank_mse[c] - rank_dm[c], | |
| } | |
| for c in movers[:12] | |
| ], | |
| } | |
| (rd / "task7_results.json").write_text(json.dumps(result, indent=2)) | |
| (rd / "config.yaml").write_text( | |
| yaml.safe_dump( | |
| { | |
| "task": 7, | |
| "manifest": str(MANIFEST), | |
| "seed": 42, | |
| "space": "prediction error (1280-d)", | |
| "maha": "full-rank error covariance (N*16 tokens)", | |
| }, | |
| sort_keys=False, | |
| ) | |
| ) | |
| np.save(rd / "error_eigenvalues.npy", lam) | |
| # plots | |
| fig, ax = plt.subplots(1, 2, figsize=(14, 5)) | |
| ax[0].semilogy(range(1, min(200, D) + 1), lam[:200], "-") | |
| ax[0].set_title( | |
| f"Error-space eigenvalue spectrum (PCA)\neff.rank {aniso['effective_rank_entropy']:.0f}/{D}, " | |
| f"top-10 var {aniso['top10_var_frac']*100:.0f}%" | |
| ) | |
| ax[0].set_xlabel("principal component") | |
| ax[0].set_ylabel("variance (eigenvalue)") | |
| ax[0].grid(alpha=0.3) | |
| xs = [rank_mse[c] for c in cats] | |
| ys = [rank_dm[c] for c in cats] | |
| ax[1].scatter(xs, ys, s=18) | |
| ax[1].plot([1, 50], [1, 50], "k--", alpha=0.5) | |
| for c in movers[:5]: | |
| ax[1].annotate(HUMAN.get(c, c), (rank_mse[c], rank_dm[c]), fontsize=8) | |
| ax[1].set_title(f"Category failure rank: MSE vs Mahalanobis\nSpearman rho={rho:.3f}") | |
| ax[1].set_xlabel("rank by Euclidean MSE (1=worst)") | |
| ax[1].set_ylabel("rank by precision-weighted D_M") | |
| ax[1].grid(alpha=0.3) | |
| fig.tight_layout() | |
| fig.savefig(rd / "task7_geometry.png", dpi=160) | |
| plt.close(fig) | |
| # console | |
| print("\n=== ANISOTROPY of the prediction-error space ===") | |
| for k, v in aniso.items(): | |
| print(f" {k}: {v:.4g}" if isinstance(v, float) else f" {k}: {v}") | |
| print("\n=== Category failure ranking: MSE vs precision-weighted Mahalanobis ===") | |
| print( | |
| f" Spearman rho = {rho:.3f} (p={pval:.1e}) (1.0 => rankings identical => insensitive to anisotropy)" | |
| ) | |
| print(" biggest rank movers (rank_mse -> rank_maha):") | |
| for m in result["biggest_rank_movers"][:8]: | |
| print( | |
| f" {m['class']} ({m['human']:>14}): {m['rank_mse']:>2} -> {m['rank_maha']:>2} (shift {m['shift']:+d})" | |
| ) | |
| print(f"\nsaved -> {rd}", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 7.41 kB
- Xet hash:
- f7077ad8d6c7054ce187071151d666204511a801d7f32eb3846d7fcd5d376faf
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.