1 files changed, 135 insertions, 0 deletions
diff --git a/worldalign/shared_rank.py b/worldalign/shared_rank.py
new file mode 100644
index 0000000..629ae0b
--- /dev/null
+++ b/worldalign/shared_rank.py
@@ -0,0 +1,135 @@
+"""Which side is narrow? Per-modality rank against the shared rank.
+
+Truncating a recovering synthetic field to rank four leaves its correlation at
+0.90 and its recovery at 6%, so the width of the shared spectrum is a separate
+and binding constraint that the correlation does not express. The engineering
+question follows immediately: which modality is narrow. Upgrading the vision
+encoder is worth its compute only if vision is what limits the shared width,
+and if the text side is the narrow one no vision backbone will help.
+
+Three numbers per field pair. Each field's own effective rank, from the
+participation ratio of its spectrum, says how many directions the modality
+uses at all. The principal angles between the two leading eigenspaces say how
+many of those directions the two modalities share. And the per-direction
+correlation profile says where along the spectrum agreement dies.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+
+import numpy as np
+import torch
+
+from .common import write_json
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--fields", nargs="+", required=True)
+    parser.add_argument("--labels", nargs="+", default=None)
+    parser.add_argument("--width", type=int, default=32)
+    parser.add_argument("--output", default="artifacts/vg_5k/shared_rank.json")
+    return parser.parse_args()
+
+
+def spectrum(matrix: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
+    symmetric = (matrix + matrix.T) / 2.0
+    values, vectors = np.linalg.eigh(symmetric)
+    order = np.argsort(np.abs(values))[::-1]
+    return values[order], vectors[:, order]
+
+
+def effective_rank(values: np.ndarray) -> float:
+    weights = np.abs(values) / np.abs(values).sum()
+    weights = weights[weights > 1e-15]
+    return float(np.exp(-(weights * np.log(weights)).sum()))
+
+
+def analyse(path: str, label: str, width: int) -> dict:
+    state = torch.load(path, map_location="cpu", weights_only=False)
+    visual = state["visual_field"].double().numpy()
+    text = state["text_field"].double().numpy()
+    size = len(visual)
+    mask = ~np.eye(size, dtype=bool)
+
+    visual_values, visual_vectors = spectrum(visual)
+    text_values, text_vectors = spectrum(text)
+    width = min(width, size)
+
+    # principal angles between the two leading eigenspaces: cosines near one
+    # count as directions the two modalities agree on. The raw count inflates
+    # with width -- two random w-dimensional subspaces of R^N already overlap
+    # at cosines near sqrt(w/N) -- so a scene-shuffled null is subtracted and
+    # every comparison is made at matched width.
+    overlap = visual_vectors[:, :width].T @ text_vectors[:, :width]
+    cosines = np.clip(np.linalg.svd(overlap, compute_uv=False), 0.0, 1.0)
+    shared = int((cosines > 0.7).sum())
+
+    null_counts = []
+    for trial in range(5):
+        generator = np.random.default_rng(trial)
+        order = generator.permutation(size)
+        shuffled = text[np.ix_(order, order)]
+        _, shuffled_vectors = spectrum(shuffled)
+        null_overlap = visual_vectors[:, :width].T @ shuffled_vectors[:, :width]
+        null_cosines = np.clip(np.linalg.svd(null_overlap, compute_uv=False), 0.0, 1.0)
+        null_counts.append(int((null_cosines > 0.7).sum()))
+    null = float(np.mean(null_counts))
+
+    # where along the spectrum does agreement die
+    profile = []
+    for index in range(min(width, 16)):
+        visual_direction = visual_vectors[:, index]
+        best = float(np.abs(text_vectors[:, :width].T @ visual_direction).max())
+        profile.append({"index": index, "best_text_overlap": best})
+
+    return {
+        "label": label,
+        "size": size,
+        "correlation": float(np.corrcoef(visual[mask], text[mask])[0, 1]),
+        "visual_effective_rank": effective_rank(visual_values),
+        "text_effective_rank": effective_rank(text_values),
+        "shared_directions_cos_gt_0.7": shared,
+        "shuffled_null": null,
+        "shared_above_null": shared - null,
+        "principal_cosines_top8": [round(float(c), 3) for c in cosines[:8]],
+        "spectrum_profile": profile,
+    }
+
+
+def main() -> None:
+    args = parse_args()
+    labels = args.labels or [path.split("/")[-1] for path in args.fields]
+    rows = [analyse(path, label, args.width)
+            for path, label in zip(args.fields, labels)]
+    for row in rows:
+        print(
+            f"{row['label']:<26} rho={row['correlation']:.3f}  "
+            f"eff_rank vision={row['visual_effective_rank']:6.1f} "
+            f"text={row['text_effective_rank']:6.1f}  "
+            f"shared={row['shared_directions_cos_gt_0.7']:3d} "
+            f"(null {row['shuffled_null']:4.1f}, net {row['shared_above_null']:5.1f})",
+            flush=True,
+        )
+    summary = {
+        "protocol": (
+            "Effective rank from the participation ratio of each field's "
+            "spectrum; shared directions from principal angles between the "
+            "two leading eigenspaces. No pairing enters beyond the alignment "
+            "the fields already carry."
+        ),
+        "width": args.width,
+        "rows": rows,
+        "reading": (
+            "The narrow modality is the one to upgrade. A shared count far "
+            "below both effective ranks means each side is rich but they are "
+            "rich about different things."
+        ),
+    }
+    write_json(args.output, summary)
+
+
+if __name__ == "__main__":
+    main()