Buckets:

affine-god
/

1111

Files

xet

affine-god/1111 / predict_l2.py

affine-god

3 months ago

download

raw

8.13 kB

"""§7 milestone: can ANYTHING predict which of two unseen thoughts scores higher?
Prefix and target answer are held fixed by construction (see microscope.py), so
a pair differs only in the thought. If a cheap text model cannot beat chance
out-of-sample here, there is no transferable semantic feature to find and the
whole semantic programme is dead BEFORE any GPU or labelling spend. If it can,
signal exists and paying to explain it is justified.
Design decisions that carry the result:
  * **Antisymmetric.** Each pair is encoded as f(A) - f(B) and mirrored to
    (B, A) with the opposite label. A model that ignores the input scores
    exactly 50% by construction, so chance is guaranteed, not assumed.
  * **Split by TURN, not by row.** Turns recur across dumps, so a random split
    would put the same prefix in train and test and inflate everything.
  * **SAME-MODEL pairs are primary.** Cross-model pairs let a ranker learn model
    identity instead of thought quality. That leak is measured at only 50.7%
    here (the king barely beats challengers), but the clean population is large
    enough that we do not need to rely on that.
  * **Near-ties are noise.** Predicting a 0.0001 Λ2 gap is hopeless and drags
    accuracy toward chance whatever the signal. Accuracy is therefore also
    reported on decisive pairs, where a real effect must show up first.
  * **Label-shuffle control.** The same pipeline on permuted labels must land at
    50%. Anything else means the split leaks and the headline is worthless.
    python harness/predict_l2.py --population same
"""

from __future__ import annotations

import argparse
import hashlib
import itertools
import json
import pathlib

import numpy as np

ROOT = pathlib.Path(__file__).resolve().parent.parent


def load_pairs(path: pathlib.Path, population: str) -> list[dict]:
    pairs = []
    for line in path.open():
        g = json.loads(line)
        if g["lp_empty_spread"] > 1e-6:      # target not truly held fixed
            continue
        for a, b in itertools.combinations(g["members"], 2):
            same = a["model"] == b["model"]
            if population == "same" and not same:
                continue
            if population == "cross" and same:
                continue
            if a["l2"] == b["l2"]:           # unorderable, carries no label
                continue
            tgt = (g["target"] or "").strip()
            pairs.append({"turn": g["turn"], "a": a["thought"], "b": b["thought"],
                          "gap": a["l2"] - b["l2"],
                          "match_a": (a["action"] or "").strip() == tgt,
                          "match_b": (b["action"] or "").strip() == tgt})
    return pairs


def split_by_turn(pairs: list[dict], frac: float = 0.3) -> tuple[list, list]:
    """Hold out whole turns. Hashed, so the split is stable across runs."""
    def held(turn: str) -> bool:
        h = hashlib.sha256(turn.encode()).digest()
        return (int.from_bytes(h[:4], "big") / 2**32) < frac
    return ([p for p in pairs if not held(p["turn"])],
            [p for p in pairs if held(p["turn"])])


def evaluate(train, test, args) -> None:
    from sklearn.feature_extraction.text import TfidfVectorizer
    from sklearn.linear_model import LogisticRegression
    from sklearn.metrics import roc_auc_score
    from scipy.sparse import vstack

    corpus = [p["a"] for p in train] + [p["b"] for p in train]
    vec = TfidfVectorizer(analyzer="word", ngram_range=(1, 2), min_df=5,
                          sublinear_tf=True, max_features=args.max_features)
    vec.fit(corpus)
    print(f"  vocabulary: {len(vec.vocabulary_)}")

    def encode(rows):
        A, B = vec.transform([r["a"] for r in rows]), vec.transform([r["b"] for r in rows])
        X = vstack([A - B, B - A])                       # mirrored -> antisymmetric
        y = np.r_[(np.array([r["gap"] for r in rows]) > 0).astype(int),
                  (np.array([r["gap"] for r in rows]) < 0).astype(int)]
        w = np.r_[np.abs([r["gap"] for r in rows]), np.abs([r["gap"] for r in rows])]
        return X, y, w

    Xtr, ytr, wtr = encode(train)
    Xte, yte, wte = encode(test)

    clf = LogisticRegression(max_iter=2000, C=args.C)
    clf.fit(Xtr, ytr)
    prob = clf.predict_proba(Xte)[:, 1]
    pred = (prob > 0.5).astype(int)

    acc = float((pred == yte).mean())
    auc = float(roc_auc_score(yte, prob))
    print(f"\n  OUT-OF-SAMPLE (held-out turns)")
    print(f"    pairs {len(yte)//2}   accuracy {acc*100:.2f}%   AUC {auc:.4f}")

    # A real effect must appear on decisive pairs before it appears anywhere.
    gaps = np.abs(np.r_[[r["gap"] for r in test], [r["gap"] for r in test]])
    print(f"\n    accuracy by |ΔΛ2| decile of the gap:")
    for lo, hi, name in ((0.5, 1.0, "top 50%"), (0.75, 1.0, "top 25%"),
                         (0.9, 1.0, "top 10%"), (0.0, 0.5, "bottom 50%")):
        t_lo, t_hi = np.quantile(gaps, lo), np.quantile(gaps, hi)
        m = (gaps >= t_lo) & (gaps <= t_hi)
        if m.sum():
            print(f"      {name:<11} |ΔΛ2| >= {t_lo:.4f}   n {int(m.sum()):>6}   "
                  f"acc {100*(pred[m] == yte[m]).mean():.2f}%")

    # In-sample, for the honest gap between what it memorised and what transfers.
    ptr = (clf.predict_proba(Xtr)[:, 1] > 0.5).astype(int)
    print(f"\n    in-sample accuracy {100*(ptr == ytr).mean():.2f}%  "
          f"(gap to out-of-sample = memorisation)")

    # Controls.
    #
    # The null must preserve the mirrored structure. Permuting the label vector
    # does NOT: it lets one pair carry label 1 on BOTH (A-B) and (B-A), which
    # leaves a fittable asymmetry and floated this control to 53%. The correct
    # null randomises each PAIR's orientation, so the text is unchanged and only
    # the direction of the comparison is random.
    rng = np.random.default_rng(0)
    n = len(train)
    flip = rng.integers(0, 2, n)
    ynull = np.r_[flip, 1 - flip]
    ctrl = LogisticRegression(max_iter=2000, C=args.C).fit(Xtr, ynull)
    cacc = float(((ctrl.predict_proba(Xte)[:, 1] > 0.5).astype(int) == yte).mean())
    se = 50.0 / np.sqrt(len(yte))
    print(f"\n  CONTROLS")
    print(f"    orientation-null accuracy {cacc*100:.2f}%   "
          f"(must be ~50% +- {2*se:.2f})")

    la = np.array([len(r["a"]) for r in test]); lb = np.array([len(r["b"]) for r in test])
    lp = np.r_[(la < lb).astype(int), (lb < la).astype(int)]   # shorter-is-better
    print(f"    length-only baseline   {100*(lp == yte).mean():.2f}%   "
          f"(known lever, worth ~8% of the bar)")


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    ap.add_argument("--data", default=str(ROOT / "microscope.jsonl"))
    ap.add_argument("--population", choices=["same", "cross", "all"], default="same")
    ap.add_argument("--max-features", type=int, default=200_000)
    ap.add_argument("--C", type=float, default=1.0)
    # Action-matching explains a +0.02318 lift but is capped at 7% of the bar by
    # the teacher's own 26.1% self-agreement. Restricting to pairs where NEITHER
    # thought reproduces the teacher's action asks whether thought-level signal
    # survives independently of that capped mechanism.
    ap.add_argument("--no-match", action="store_true",
                    help="keep only pairs where neither action equals the target")
    args = ap.parse_args()

    pairs = load_pairs(pathlib.Path(args.data), args.population)
    if args.no_match:
        before = len(pairs)
        pairs = [p for p in pairs if not p["match_a"] and not p["match_b"]]
        print(f"restricted to non-matching pairs: {before} -> {len(pairs)}")
    train, test = split_by_turn(pairs)
    tr_turns = {p["turn"] for p in train}; te_turns = {p["turn"] for p in test}
    print(f"population '{args.population}': {len(pairs)} orderable pairs")
    print(f"  train {len(train)} pairs / {len(tr_turns)} turns   "
          f"test {len(test)} pairs / {len(te_turns)} turns")
    assert not (tr_turns & te_turns), "turn leaked across the split"
    evaluate(train, test, args)


if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 8.13 kB
Xet hash:: 1263f4176e145d079c074e084f5799b5c50d9d37bb70592681111790025fb227

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""§7 milestone: can ANYTHING predict which of two unseen thoughts scores higher?

	Prefix and target answer are held fixed by construction (see microscope.py), so
	a pair differs only in the thought. If a cheap text model cannot beat chance
	out-of-sample here, there is no transferable semantic feature to find and the
	whole semantic programme is dead BEFORE any GPU or labelling spend. If it can,
	signal exists and paying to explain it is justified.

	Design decisions that carry the result:

	* Antisymmetric. Each pair is encoded as f(A) - f(B) and mirrored to
	(B, A) with the opposite label. A model that ignores the input scores
	exactly 50% by construction, so chance is guaranteed, not assumed.

	* Split by TURN, not by row. Turns recur across dumps, so a random split
	would put the same prefix in train and test and inflate everything.

	* SAME-MODEL pairs are primary. Cross-model pairs let a ranker learn model
	identity instead of thought quality. That leak is measured at only 50.7%
	here (the king barely beats challengers), but the clean population is large
	enough that we do not need to rely on that.

	* Near-ties are noise. Predicting a 0.0001 Λ2 gap is hopeless and drags
	accuracy toward chance whatever the signal. Accuracy is therefore also
	reported on decisive pairs, where a real effect must show up first.

	* Label-shuffle control. The same pipeline on permuted labels must land at
	50%. Anything else means the split leaks and the headline is worthless.

	python harness/predict_l2.py --population same
	"""

	from __future__ import annotations

	import argparse
	import hashlib
	import itertools
	import json
	import pathlib

	import numpy as np

	ROOT = pathlib.Path(__file__).resolve().parent.parent


	def load_pairs(path: pathlib.Path, population: str) -> list[dict]:
	pairs = []
	for line in path.open():
	g = json.loads(line)
	if g["lp_empty_spread"] > 1e-6: # target not truly held fixed
	continue
	for a, b in itertools.combinations(g["members"], 2):
	same = a["model"] == b["model"]
	if population == "same" and not same:
	continue
	if population == "cross" and same:
	continue
	if a["l2"] == b["l2"]: # unorderable, carries no label
	continue
	tgt = (g["target"] or "").strip()
	pairs.append({"turn": g["turn"], "a": a["thought"], "b": b["thought"],
	"gap": a["l2"] - b["l2"],
	"match_a": (a["action"] or "").strip() == tgt,
	"match_b": (b["action"] or "").strip() == tgt})
	return pairs


	def split_by_turn(pairs: list[dict], frac: float = 0.3) -> tuple[list, list]:
	"""Hold out whole turns. Hashed, so the split is stable across runs."""
	def held(turn: str) -> bool:
	h = hashlib.sha256(turn.encode()).digest()
	return (int.from_bytes(h[:4], "big") / 2**32) < frac
	return ([p for p in pairs if not held(p["turn"])],
	[p for p in pairs if held(p["turn"])])


	def evaluate(train, test, args) -> None:
	from sklearn.feature_extraction.text import TfidfVectorizer
	from sklearn.linear_model import LogisticRegression
	from sklearn.metrics import roc_auc_score
	from scipy.sparse import vstack

	corpus = [p["a"] for p in train] + [p["b"] for p in train]
	vec = TfidfVectorizer(analyzer="word", ngram_range=(1, 2), min_df=5,
	sublinear_tf=True, max_features=args.max_features)
	vec.fit(corpus)
	print(f" vocabulary: {len(vec.vocabulary_)}")

	def encode(rows):
	A, B = vec.transform([r["a"] for r in rows]), vec.transform([r["b"] for r in rows])
	X = vstack([A - B, B - A]) # mirrored -> antisymmetric
	y = np.r_[(np.array([r["gap"] for r in rows]) > 0).astype(int),
	(np.array([r["gap"] for r in rows]) < 0).astype(int)]
	w = np.r_[np.abs([r["gap"] for r in rows]), np.abs([r["gap"] for r in rows])]
	return X, y, w

	Xtr, ytr, wtr = encode(train)
	Xte, yte, wte = encode(test)

	clf = LogisticRegression(max_iter=2000, C=args.C)
	clf.fit(Xtr, ytr)
	prob = clf.predict_proba(Xte)[:, 1]
	pred = (prob > 0.5).astype(int)

	acc = float((pred == yte).mean())
	auc = float(roc_auc_score(yte, prob))
	print(f"\n OUT-OF-SAMPLE (held-out turns)")
	print(f" pairs {len(yte)//2} accuracy {acc*100:.2f}% AUC {auc:.4f}")

	# A real effect must appear on decisive pairs before it appears anywhere.
	gaps = np.abs(np.r_[[r["gap"] for r in test], [r["gap"] for r in test]])
	print(f"\n accuracy by \|ΔΛ2\| decile of the gap:")
	for lo, hi, name in ((0.5, 1.0, "top 50%"), (0.75, 1.0, "top 25%"),
	(0.9, 1.0, "top 10%"), (0.0, 0.5, "bottom 50%")):
	t_lo, t_hi = np.quantile(gaps, lo), np.quantile(gaps, hi)
	m = (gaps >= t_lo) & (gaps <= t_hi)
	if m.sum():
	print(f" {name:<11} \|ΔΛ2\| >= {t_lo:.4f} n {int(m.sum()):>6} "
	f"acc {100*(pred[m] == yte[m]).mean():.2f}%")

	# In-sample, for the honest gap between what it memorised and what transfers.
	ptr = (clf.predict_proba(Xtr)[:, 1] > 0.5).astype(int)
	print(f"\n in-sample accuracy {100*(ptr == ytr).mean():.2f}% "
	f"(gap to out-of-sample = memorisation)")

	# Controls.
	#
	# The null must preserve the mirrored structure. Permuting the label vector
	# does NOT: it lets one pair carry label 1 on BOTH (A-B) and (B-A), which
	# leaves a fittable asymmetry and floated this control to 53%. The correct
	# null randomises each PAIR's orientation, so the text is unchanged and only
	# the direction of the comparison is random.
	rng = np.random.default_rng(0)
	n = len(train)
	flip = rng.integers(0, 2, n)
	ynull = np.r_[flip, 1 - flip]
	ctrl = LogisticRegression(max_iter=2000, C=args.C).fit(Xtr, ynull)
	cacc = float(((ctrl.predict_proba(Xte)[:, 1] > 0.5).astype(int) == yte).mean())
	se = 50.0 / np.sqrt(len(yte))
	print(f"\n CONTROLS")
	print(f" orientation-null accuracy {cacc*100:.2f}% "
	f"(must be ~50% +- {2*se:.2f})")

	la = np.array([len(r["a"]) for r in test]); lb = np.array([len(r["b"]) for r in test])
	lp = np.r_[(la < lb).astype(int), (lb < la).astype(int)] # shorter-is-better
	print(f" length-only baseline {100*(lp == yte).mean():.2f}% "
	f"(known lever, worth ~8% of the bar)")


	def main() -> None:
	ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
	ap.add_argument("--data", default=str(ROOT / "microscope.jsonl"))
	ap.add_argument("--population", choices=["same", "cross", "all"], default="same")
	ap.add_argument("--max-features", type=int, default=200_000)
	ap.add_argument("--C", type=float, default=1.0)
	# Action-matching explains a +0.02318 lift but is capped at 7% of the bar by
	# the teacher's own 26.1% self-agreement. Restricting to pairs where NEITHER
	# thought reproduces the teacher's action asks whether thought-level signal
	# survives independently of that capped mechanism.
	ap.add_argument("--no-match", action="store_true",
	help="keep only pairs where neither action equals the target")
	args = ap.parse_args()

	pairs = load_pairs(pathlib.Path(args.data), args.population)
	if args.no_match:
	before = len(pairs)
	pairs = [p for p in pairs if not p["match_a"] and not p["match_b"]]
	print(f"restricted to non-matching pairs: {before} -> {len(pairs)}")
	train, test = split_by_turn(pairs)
	tr_turns = {p["turn"] for p in train}; te_turns = {p["turn"] for p in test}
	print(f"population '{args.population}': {len(pairs)} orderable pairs")
	print(f" train {len(train)} pairs / {len(tr_turns)} turns "
	f"test {len(test)} pairs / {len(te_turns)} turns")
	assert not (tr_turns & te_turns), "turn leaked across the split"
	evaluate(train, test, args)


	if __name__ == "__main__":
	main()