#!/usr/bin/env python3 """house_metrics.py — P-A, H1-HOUSE and the §3.1(7) near-duplicate scan. The instrument is **not rebuilt**: every estimator, seed, draw count and duplicate-pair rule is imported from the SEALED `run_bench.py` pinned by `PREREG-COHERENCE.md` Amendment A1 (`PREREG-HOUSE-ARMS.md` §8). The mnml corpora's archived embeddings are read at their pinned sha and **never re-cut** (§8.3) — re-cutting them would re-open the one thing that makes 0.4489 comparable. What it computes, each registered before any number existed: * **STEP 0′ (§9.2)** — the reproduction gate. mnml's equal-N=19 row is re-derived from the archived embeddings in the RNG position it occupied (fifth of five, `CORPUS_ORDER`), and the run aborts unless it reproduces `0.448852` / `[0.365211, 0.523948]` to six decimals. * **P-A (§9.2)** — the house training set's equal-N=19 `D_pair` and its 95 % bootstrap CI, for the THREE registered sets, with the registered decision rule (holds iff the CI lies entirely below mnml's; fails iff entirely above; otherwise NOT DETECTED — never an equivalence). Primary stream: the house corpus draws **fresh at position 1**. Sensitivity: the house CI under mnml's fifth-position stream, so the stream cannot be the thing that decides. * **H1-HOUSE (§8.2)** — `D_pair(house-wide train) > D_pair(house-artist train)` at N = 19 primary and N_min secondary; non-overlapping 95 % CIs primary, one-sided permutation p secondary; a disagreement between them IS the finding. * **§3.1(7)** — the cross-split near-duplicate scan. Every (holdout, training) pair below the 1st percentile of the training set's OWN pairwise distance distribution is named, so the split can be called a *work* split rather than only an *artist* split. MEASURED, NEVER GATING (D-20260831-17). """ from __future__ import annotations import argparse import csv import importlib.util import json import sys from pathlib import Path # §9.2 — the comparator is READ, not recomputed. MNML_DPAIR = 0.448852 MNML_CI = (0.365211, 0.523948) ARM3_ARTIST = "artist_487628" def load_module(path: Path, name: str): spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) sys.modules[name] = mod spec.loader.exec_module(mod) return mod def track_vectors(V, idx, pred): """A track's vector = the L2-normalised mean of its three crops (§4.2).""" import numpy as np buckets: dict[str, list[int]] = {} for r in idx: if pred(r): buckets.setdefault(r["source_sha256"], []).append(int(r["row"])) out = {} for sha, rows in buckets.items(): v = V[sorted(rows)].mean(axis=0) out[sha] = v / np.linalg.norm(v) return out def summarise(rb, E, rng_sub, rng_boot): """One corpus's row, using the sealed estimators and the sealed draw counts.""" import numpy as np dp_eq, _ = rb.equal_n_draws(E, rng_sub) dp_bs, _ = rb.bootstrap_draws(E, rng_boot) return { "n": int(len(E)), "d_pair_full": round(rb.d_pair(E), 6), "d_cent_full": round(rb.d_cent(E), 6), "d_pair_equaln": round(float(dp_eq.mean()), 6), "d_pair_equaln_sd": round(float(dp_eq.std(ddof=0)), 6), "d_pair_ci_lo": round(float(np.percentile(dp_bs, 2.5)), 6), "d_pair_ci_hi": round(float(np.percentile(dp_bs, 97.5)), 6), } def pa_verdict(ci_lo: float, ci_hi: float) -> str: """§9.2's registered decision rule. Three outcomes, exhaustive and exclusive.""" if ci_hi < MNML_CI[0]: return "HOLDS" if ci_lo > MNML_CI[1]: return "FAILS" return "NOT DETECTED" def permutation_one_sided(rb, E_a, E_b, label_a, label_b): """§5.4's estimator: is D_pair(a) - D_pair(b) larger than chance? one-sided.""" import numpy as np pooled = np.concatenate([E_a, E_b]) G = pooled @ pooled.T n_a = len(E_a) def eq_dpair(members, rng): members = np.asarray(members) if len(members) <= rb.N_EQUAL: sub = G[np.ix_(members, members)] iu = np.triu_indices(len(members), k=1) return float((1.0 - sub[iu]).mean()) picks = np.array([rng.choice(members, rb.N_EQUAL, replace=False) for _ in range(rb.N_DRAWS)]) sub = G[picks[:, :, None], picks[:, None, :]] iu = np.triu_indices(rb.N_EQUAL, k=1) return float((1.0 - sub[:, iu[0], iu[1]]).mean()) rng_obs = np.random.default_rng(rb.SEED_SUBSAMPLE) obs = eq_dpair(np.arange(n_a), rng_obs) - eq_dpair(np.arange(n_a, len(pooled)), rng_obs) rng_p = np.random.default_rng(rb.SEED_PERMUTE) all_idx = np.arange(len(pooled)) ge = 0 for _ in range(rb.N_PERM): perm = rng_p.permutation(all_idx) stat = eq_dpair(perm[:n_a], rng_p) - eq_dpair(perm[n_a:], rng_p) if stat >= obs: ge += 1 return { "comparison": f"D_pair({label_a}) - D_pair({label_b})", "observed": round(obs, 6), "n_permutations": rb.N_PERM, "p_one_sided": round((ge + 1) / (rb.N_PERM + 1), 6), "note": ("one-sided, testing whether the FIRST set is more spread than the " "second; p is (ge + 1) / (n_perm + 1), the conservative form"), } def main() -> int: import numpy as np ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--bench", type=Path, default=Path.home() / "music/out/coherence-bench/run_bench.py") ap.add_argument("--bench-sha", required=True) ap.add_argument("--archive", type=Path, default=Path.home() / "music/out/coherence-bench", help="the mnml arc's archived embeddings — READ ONLY, never re-cut (§8.3)") ap.add_argument("--house", type=Path, default=Path.home() / "music/out/house-coherence") ap.add_argument("--split", type=Path, required=True) ap.add_argument("--manifest", type=Path, required=True) ap.add_argument("--split-rule", type=Path, required=True, help="split_rule.py — arm 3's by-track split comes from the same module") ap.add_argument("--out", type=Path, required=True) args = ap.parse_args() import hashlib got = hashlib.sha256(args.bench.read_bytes()).hexdigest() if got != args.bench_sha: print(f"ABORT: sealed bench hashes to {got}", file=sys.stderr) return 2 rb = load_module(args.bench, "run_bench_pinned") sr = load_module(args.split_rule, "split_rule_pinned") print(f"OK sealed instrument {args.bench.name} = {got}") print(f"OK N_EQUAL={rb.N_EQUAL} N_DRAWS={rb.N_DRAWS} N_PERM={rb.N_PERM} " f"seeds={rb.SEED_SUBSAMPLE}/{rb.SEED_BOOTSTRAP}/{rb.SEED_PERMUTE}") results: dict = { "generated_utc": __import__("time").strftime("%Y-%m-%dT%H:%M:%SZ", __import__("time").gmtime()), "instrument": { "run_bench_sha256": got, "coherence_prereg_seal": rb.SEAL, "clap_ckpt_sha256": rb.CKPT_SHA, "freeze_A1_sha256": rb.FREEZE_SHA, "n_equal": rb.N_EQUAL, "n_draws": rb.N_DRAWS, "n_perm": rb.N_PERM, "seeds": {"subsample": rb.SEED_SUBSAMPLE, "bootstrap": rb.SEED_BOOTSTRAP, "permute": rb.SEED_PERMUTE}, "corpus_rng_order": list(rb.CORPUS_ORDER), }, "standing_constraint": "MEASURED, NEVER GATING (D-20260831-17)", } # ---- the archived mnml-arc embeddings, at their pinned sha (§8.3) ------- arc_npz, arc_idx_csv = args.archive / "embeddings.npz", args.archive / "embeddings-index.csv" arc_shas = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in (arc_npz, arc_idx_csv)} print(f"OK archived embeddings.npz = {arc_shas['embeddings.npz']}") print(f"OK archived embeddings-index.csv = {arc_shas['embeddings-index.csv']}") results["archive_shas"] = arc_shas Va = np.load(arc_npz)["crops"] ia = list(csv.DictReader(open(arc_idx_csv))) primary = (lambda c: (lambda r: r["set"] == "corpus" and r["mode"] == "norm" and r["in_train"] == "1" and r["corpus"] == c)) arc_tv = {c: track_vectors(Va, ia, primary(c)) for c in rb.CORPUS_ORDER} for c in rb.CORPUS_ORDER: print(f"OK archived {c:<11} training tracks = {len(arc_tv[c])}") # ---- STEP 0' — THE REPRODUCTION GATE (§9.2) ---------------------------- rng_sub = np.random.default_rng(rb.SEED_SUBSAMPLE) rng_boot = np.random.default_rng(rb.SEED_BOOTSTRAP) arc_rows = {} for c in rb.CORPUS_ORDER: # consume the streams IN ORDER E = np.stack([arc_tv[c][s] for s in sorted(arc_tv[c])]) arc_rows[c] = summarise(rb, E, rng_sub, rng_boot) m = arc_rows["mnml"] print(f"\nSTEP 0' mnml equal-N=19 D_pair = {m['d_pair_equaln']:.6f} " f"CI [{m['d_pair_ci_lo']:.6f}, {m['d_pair_ci_hi']:.6f}]") print(f"STEP 0' REQUIRES {MNML_DPAIR:.6f} " f"CI [{MNML_CI[0]:.6f}, {MNML_CI[1]:.6f}]") if (round(m["d_pair_equaln"], 6) != MNML_DPAIR or round(m["d_pair_ci_lo"], 6) != MNML_CI[0] or round(m["d_pair_ci_hi"], 6) != MNML_CI[1]): print("STEP 0' ABORT — the archived row was not reproduced to six decimals", file=sys.stderr) return 2 print("STEP 0' PASS — the comparator reproduces exactly; the house sets may be read.\n") results["step0_prime"] = {"status": "PASS", "mnml_row": m, "required": {"d_pair_equaln": MNML_DPAIR, "ci": list(MNML_CI)}} results["archived_rows"] = arc_rows # ---- the house sets ---------------------------------------------------- Vh = np.load(args.house / "embeddings.npz")["crops"] ih = list(csv.DictReader(open(args.house / "embeddings-index.csv"))) house_norm = track_vectors(Vh, ih, lambda r: r["mode"] == "norm") print(f"OK house tracks embedded (norm) = {len(house_norm)}") split = json.loads(args.split.read_text("utf-8")) manifest = [json.loads(l) for l in args.manifest.read_text("utf-8").splitlines() if l.strip()] sha_of = {r["track_id"]: r["local_sha256"] for r in manifest} artist_of = {r["track_id"]: r["artist_id"] for r in manifest} title_of = {r["track_id"]: r["title"] for r in manifest} name_of = {r["track_id"]: r["artist_name"] for r in manifest} train_ids = sorted(t for a in split["train"]["artists"] for t in a["track_ids"]) hold_ids = sorted(t for a in split["held_out"]["artists"] for t in a["track_ids"]) arm3_rows = [r for r in manifest if r["artist_id"] == ARM3_ARTIST] arm3 = sr.by_track_split(arm3_rows) print(f"OK arm-3 corpus ({ARM3_ARTIST}) = {len(arm3_rows)} tracks -> " f"{arm3.n_train} train / {arm3.n_holdout} held (§3.3, seed 42)") def E_of(ids): return np.stack([house_norm[sha_of[t]] for t in ids]) SETS = { "house_wide_train": (train_ids, "THE HEADLINE — arm 1's frozen TRAINING set (§9.2(i))"), "house_wide_pool": (train_ids + hold_ids, "the full frozen pool, 416 tracks (§9.2(ii))"), "house_wide_train_without_arm3_artist": ( [t for t in train_ids if artist_of[t] != ARM3_ARTIST], "the training set with artist_487628 REMOVED — §9.2(iii) registered " "'with the artist re-included', which under the D-20260905-95 ruling " "of ⧖ H-1 as option (b) is already set (i); the informative complement " "is therefore its removal, and the substitution is named rather than " "silently swapped"), "house_wide_holdout": (hold_ids, "the held-out side, descriptive"), "house_artist_train": (list(arm3.training_track_ids), "ARM 3's training set (§13.2), nested inside arm 1"), } # PRIMARY: the house corpus draws as a FRESH stream at position 1 (§9.2). print("\n--- P-A, primary: the house sets draw FRESH at RNG position 1 ---") house_rows = {} for key, (ids, desc) in SETS.items(): rs, rbo = (np.random.default_rng(rb.SEED_SUBSAMPLE), np.random.default_rng(rb.SEED_BOOTSTRAP)) row = summarise(rb, E_of(ids), rs, rbo) row["description"] = desc house_rows[key] = row print(f" {key:<40} n={row['n']:>3} D_pair(N=19)={row['d_pair_equaln']:.6f} " f"CI [{row['d_pair_ci_lo']:.6f}, {row['d_pair_ci_hi']:.6f}] " f"full={row['d_pair_full']:.6f}") # SENSITIVITY: the same set drawn in mnml's FIFTH-position stream. print("\n--- P-A, sensitivity: the same sets drawn in mnml's FIFTH-position stream ---") sens = {} for key, (ids, _) in SETS.items(): rs, rbo = (np.random.default_rng(rb.SEED_SUBSAMPLE), np.random.default_rng(rb.SEED_BOOTSTRAP)) for c in rb.CORPUS_ORDER[:4]: # burn positions 1-4 exactly as the arc did Ec = np.stack([arc_tv[c][s] for s in sorted(arc_tv[c])]) rb.equal_n_draws(Ec, rs) rb.bootstrap_draws(Ec, rbo) row = summarise(rb, E_of(ids), rs, rbo) sens[key] = row print(f" {key:<40} D_pair(N=19)={row['d_pair_equaln']:.6f} " f"CI [{row['d_pair_ci_lo']:.6f}, {row['d_pair_ci_hi']:.6f}]") head = house_rows["house_wide_train"] verdict = pa_verdict(head["d_pair_ci_lo"], head["d_pair_ci_hi"]) verdict_sens = pa_verdict(sens["house_wide_train"]["d_pair_ci_lo"], sens["house_wide_train"]["d_pair_ci_hi"]) print(f"\nP-A PRIMARY VERDICT: {verdict} " f"(house CI [{head['d_pair_ci_lo']:.6f}, {head['d_pair_ci_hi']:.6f}] " f"vs mnml CI [{MNML_CI[0]:.6f}, {MNML_CI[1]:.6f}])") print(f"P-A under the fifth-position stream: {verdict_sens}") perm_pa = permutation_one_sided( rb, E_of(train_ids), np.stack([arc_tv["mnml"][s] for s in sorted(arc_tv["mnml"])]), "house-wide train", "mnml train") # P-A predicts house < mnml, so the registered one-sided direction is the # complement of the "house more spread" statistic this estimator computes. perm_pa["p_one_sided_house_less_spread"] = round(1.0 - perm_pa["p_one_sided"], 6) print(f"P-A secondary permutation: observed {perm_pa['observed']:+.6f}, " f"p(house MORE spread) = {perm_pa['p_one_sided']:.4f}, " f"p(house LESS spread) = {perm_pa['p_one_sided_house_less_spread']:.4f}") results["P_A"] = { "claim": ("the house-wide TRAINING set's mean pairwise CLAP cosine distance " "lands BELOW the minimal-techno training set's, same instrument, same N"), "prediction": "YES, confidence moderate (registered before any number existed)", "comparator_read_not_recomputed": {"d_pair_equaln": MNML_DPAIR, "ci": list(MNML_CI), "source": "music/out/coherence-bench/" "out/metric1-corpus-spread.csv row primary,mnml"}, "primary_rng_position": 1, "primary": house_rows, "sensitivity_fifth_position_stream": sens, "verdict_primary": verdict, "verdict_under_fifth_position_stream": verdict_sens, "secondary_permutation": perm_pa, "decision_rule": ("HOLDS iff the house CI lies entirely below mnml's; FAILS iff " "entirely above; otherwise NOT DETECTED — never an equivalence"), } # ---- H1-HOUSE (§8.2) --------------------------------------------------- wide, artist = house_rows["house_wide_train"], house_rows["house_artist_train"] overlap = not (wide["d_pair_ci_lo"] > artist["d_pair_ci_hi"] or artist["d_pair_ci_lo"] > wide["d_pair_ci_hi"]) perm_h1 = permutation_one_sided(rb, E_of(train_ids), E_of(list(arm3.training_track_ids)), "house-wide train", "house-artist train") h1_primary = ("SUPPORTED" if (not overlap and wide["d_pair_equaln"] > artist["d_pair_equaln"]) else "NOT SUPPORTED") h1_secondary = "SUPPORTED" if perm_h1["p_one_sided"] < 0.05 else "NOT SUPPORTED" print(f"\nH1-HOUSE wide {wide['d_pair_equaln']:.6f} " f"[{wide['d_pair_ci_lo']:.6f}, {wide['d_pair_ci_hi']:.6f}] vs " f"artist {artist['d_pair_equaln']:.6f} " f"[{artist['d_pair_ci_lo']:.6f}, {artist['d_pair_ci_hi']:.6f}]") print(f"H1-HOUSE primary (non-overlapping CIs): {h1_primary} " f"secondary (permutation p={perm_h1['p_one_sided']:.4f}): {h1_secondary}") results["H1_HOUSE"] = { "hypothesis": "D_pair(house-wide train) > D_pair(house-artist train)", "prediction": "yes, confidence high", "n_axis_primary": rb.N_EQUAL, "wide": wide, "artist": artist, "cis_overlap": overlap, "verdict_primary_non_overlapping_ci": h1_primary, "verdict_secondary_permutation": h1_secondary, "permutation": perm_h1, "agree": h1_primary == h1_secondary, "nesting_named": ("arm 3's 37 training tracks are a SUBSET of arm 1's 331 " "under the D-20260905-95 ruling of ⧖ H-1 as (b); this is " "'the same 37 tracks with and without 294 others', not two " "independent corpora"), "mnml_arc_comparator": {"wide": MNML_DPAIR, "one_artist": 0.2378}, } # ---- §3.1(7) THE CROSS-SPLIT NEAR-DUPLICATE SCAN ----------------------- Et, Eh = E_of(train_ids), E_of(hold_ids) Gtt = Et @ Et.T iu = np.triu_indices(len(Et), k=1) train_dists = 1.0 - Gtt[iu] thresh = float(np.percentile(train_dists, 1.0)) Dth = 1.0 - (Eh @ Et.T) flagged = [] for i, j in zip(*np.where(Dth < thresh)): flagged.append({ "holdout_track": hold_ids[i], "holdout_title": title_of[hold_ids[i]], "holdout_artist": name_of[hold_ids[i]], "training_track": train_ids[j], "training_title": title_of[train_ids[j]], "training_artist": name_of[train_ids[j]], "distance": round(float(Dth[i, j]), 6), }) flagged.sort(key=lambda x: x["distance"]) print(f"\n§3.1(7) training-set pairwise distances: n={len(train_dists)}, " f"1st percentile = {thresh:.6f}") print(f"§3.1(7) (holdout, training) pairs below it: {len(flagged)}") for f in flagged[:15]: print(f" {f['distance']:.6f} {f['holdout_track']} " f"\"{f['holdout_title']}\" ({f['holdout_artist']}) ~ " f"{f['training_track']} \"{f['training_title']}\" ({f['training_artist']})") results["cross_split_near_duplicate_scan"] = { "rule": ("every (holdout, training) pair below the 1st percentile of the " "TRAINING set's own pairwise distance distribution is named, " "listened to, and if it is the same recording the holdout copy " "moves to training or both are excluded under E5"), "training_pairs": int(len(train_dists)), "percentile": 1.0, "threshold_distance": round(thresh, 6), "dataset_hash_source": str(args.manifest), "flagged_pairs": len(flagged), "pairs": flagged, "consequence": ("absent this scan the split is an ARTIST split, not a WORK " "split; with it, every flagged pair is named here"), } args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(results, indent=2, ensure_ascii=False) + "\n", "utf-8") print(f"\nWROTE {args.out}") return 0 if __name__ == "__main__": raise SystemExit(main())