Upload folder using huggingface_hub

18104f9 verified about 2 months ago

12.2 kB

#!/usr/bin/env python3
"""Compare every model across the four evaluation splits.
    python analysis/dataset_level_analysis/analyze_dataset_level.py
Reads result/<model>/<task>/baseline_<b>/{metrics.json,predictions.jsonl} and
writes the CSVs next to this script.
Raw accuracy is not comparable across the four splits: VSR is a two-way
true/false question while the three Youtube levels are four-way, and level 2
carries a structural floor of its own (see FLOORS).  Everything is therefore
also reported as *headroom captured*, (acc - floor) / (1 - floor): the share of
the distance between guessing and perfect that the model actually covers.  It
is negative when a model does worse than the floor, which happens here.
"""
from __future__ import annotations

import argparse
import collections
import csv
import json
import math
import os
import statistics

BASELINES = ["0.00", "0.05", "0.10", "0.15", "0.20", "0.25"]
TASKS = ["vsr", "youtube_level1", "youtube_level2", "youtube_level3"]

# What a model scores on each split without looking at the image: pure guessing
# among the choices.  Level 2's three distractors are drawn uniformly from the
# 23 wrong orderings, so the four options are exchangeable and every text-only
# heuristic sits at chance -- measured on the split: prefix vote 0.252, medoid
# 0.256, outlier 0.254, positional majority 0.245.  Earlier structured variants
# did NOT have this property; see disparity_qa/level2_choices.py.
FLOORS = {"vsr": 0.500, "youtube_level1": 0.250,
          "youtube_level2": 0.250, "youtube_level3": 0.250}
CHANCE = dict(FLOORS)


def load_metrics(root: str, model: str, task: str, baseline: str) -> dict:
    with open(os.path.join(root, "result", model, task,
                           f"baseline_{baseline}", "metrics.json")) as fh:
        return json.load(fh)


def load_predictions(root: str, model: str, task: str, baseline: str):
    path = os.path.join(root, "result", model, task,
                        f"baseline_{baseline}", "predictions.jsonl")
    with open(path) as fh:
        for line in fh:
            line = line.strip()
            if line:
                yield json.loads(line)


def entropy(counter: collections.Counter) -> float:
    n = sum(counter.values())
    if not n:
        return 0.0
    return -sum((v / n) * math.log2(v / n) for v in counter.values() if v)


def slope(xs, ys) -> float:
    """Least-squares slope of accuracy against baseline (per unit of baseline)."""
    mx, my = statistics.mean(xs), statistics.mean(ys)
    den = sum((x - mx) ** 2 for x in xs)
    return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / den if den else 0.0


def main() -> int:
    here = os.path.dirname(os.path.abspath(__file__))
    p = argparse.ArgumentParser()
    p.add_argument("--root", default=os.path.abspath(os.path.join(here, "..", "..")))
    p.add_argument("--out_dir", default=here)
    args = p.parse_args()

    models = sorted(m for m in os.listdir(os.path.join(args.root, "result"))
                    if not m.startswith("_")
                    and os.path.isdir(os.path.join(args.root, "result", m)))

    acc = {}          # (model, task, baseline) -> metrics dict
    for m in models:
        for t in TASKS:
            for b in BASELINES:
                acc[(m, t, b)] = load_metrics(args.root, m, t, b)

    # ---------------------------------------------------------- per baseline
    rows = []
    for m in models:
        for t in TASKS:
            for b in BASELINES:
                d = acc[(m, t, b)]
                rows.append({
                    "model": m, "task": t, "baseline": b,
                    "n": d["num_samples"], "accuracy": round(d["accuracy"], 4),
                    "num_correct": d["num_correct"],
                    "unparsed_rate": round(d["unparsed_rate"], 4),
                    "scoring": d["scoring"],
                    "chance": CHANCE[t], "floor": FLOORS[t],
                    "headroom_captured": round(
                        (d["accuracy"] - FLOORS[t]) / (1 - FLOORS[t]), 4),
                })
    write_csv(os.path.join(args.out_dir, "accuracy_by_baseline.csv"), rows)

    # ------------------------------------------------------- per model/task
    summary = []
    for m in models:
        for t in TASKS:
            a = [acc[(m, t, b)]["accuracy"] for b in BASELINES]
            xs = [float(b) for b in BASELINES]
            summary.append({
                "model": m, "task": t, "n": acc[(m, t, "0.00")]["num_samples"],
                "scoring": acc[(m, t, "0.00")]["scoring"],
                "chance": CHANCE[t], "floor": FLOORS[t],
                "acc_b0": round(a[0], 4), "acc_mean": round(statistics.mean(a), 4),
                "acc_min": round(min(a), 4), "acc_max": round(max(a), 4),
                "acc_range": round(max(a) - min(a), 4),
                "acc_std": round(statistics.pstdev(a), 4),
                "slope_per_baseline": round(slope(xs, a), 4),
                "delta_b025_minus_b0": round(a[-1] - a[0], 4),
                "headroom_captured_mean": round(
                    (statistics.mean(a) - FLOORS[t]) / (1 - FLOORS[t]), 4),
                "above_floor": statistics.mean(a) > FLOORS[t],
            })
    write_csv(os.path.join(args.out_dir, "summary_by_model_task.csv"), summary)

    # --------------------------------------------------------------- biases
    bias = []
    for m in models:
        for t in TASKS:
            for b in BASELINES:
                pred = collections.Counter()
                ref = collections.Counter()
                n = 0
                for r in load_predictions(args.root, m, t, b):
                    pred[r["prediction_norm"]] += 1
                    ref[r["reference_norm"]] += 1
                    n += 1
                top_label, top_n = pred.most_common(1)[0]
                # what a model would score by always naming its own modal answer
                majority_acc = ref[top_label] / n if n else 0.0
                bias.append({
                    "model": m, "task": t, "baseline": b, "n": n,
                    "n_distinct_predictions": len(pred),
                    "modal_prediction": top_label,
                    "modal_share": round(top_n / n, 4),
                    "prediction_entropy_bits": round(entropy(pred), 4),
                    "reference_entropy_bits": round(entropy(ref), 4),
                    "always_modal_accuracy": round(majority_acc, 4),
                    "accuracy": round(acc[(m, t, b)]["accuracy"], 4),
                    "prediction_counts": json.dumps(dict(sorted(pred.items()))),
                })
    write_csv(os.path.join(args.out_dir, "prediction_bias.csv"), bias)

    ladder = level2_ladder(args.root, models)
    if ladder:
        write_csv(os.path.join(args.out_dir, "level2_structure_use.csv"), ladder)

    report(models, acc, summary, bias, ladder)
    print(f"\nwrote {3 + bool(ladder)} CSVs to {args.out_dir}")
    return 0


# the split lives in the repo, one level above this experiment directory
LEVEL2_TEST = os.path.join("..", "02_data", "Youtube_self_depth_QA",
                           "level2", "test.jsonl")


def _prefix_survivors(options):
    """The options a majority vote on successive positions cannot rule out.
    A regression guard.  With random distractors this should leave most items
    undecided and score at chance; if it ever climbs toward 0.5 again, the
    option construction has reacquired a structural leak.
    """
    alive = list(options)
    seqs = {lab: txt.split(" > ") for lab, txt in options.items()}
    for k in range(4):
        if len(alive) <= 1:
            break
        counts = collections.Counter(seqs[a][k] for a in alive)
        top = counts.most_common(1)[0][1]
        winners = [l for l, v in counts.items() if v == top]
        if len(winners) == 1:
            alive = [a for a in alive if seqs[a][k] == winners[0]]
    return alive


def level2_ladder(root: str, models):
    """How often each model's answer lands inside the structure-only survivor set."""
    path = os.path.abspath(os.path.join(root, LEVEL2_TEST))
    if not os.path.exists(path):
        print(f"[warn] level-2 split not found at {path}; skipping ladder analysis")
        return []
    test = {}
    with open(path) as fh:
        for line in fh:
            rec = json.loads(line)
            if rec.get("options"):
                test[rec["sample_id"]] = rec["options"]
    if not test:
        return []
    out = []
    for m in models:
        for b in BASELINES:
            inside = hit = n = 0
            for r in load_predictions(root, m, "youtube_level2", b):
                opts = test.get(r["sample_id"])
                if not opts:
                    continue
                alive = _prefix_survivors(opts)
                n += 1
                inside += int(r["prediction_norm"] in alive)
                hit += int(r["correct"])
            out.append({
                "model": m, "baseline": b, "n": n,
                "answer_in_survivor_set": round(inside / n, 4) if n else 0.0,
                "accuracy": round(hit / n, 4) if n else 0.0,
                # of the answers that did land in the survivor set, how many were
                # right: this is the coin flip the structure cannot decide
                "accuracy_within_survivors": round(hit / inside, 4) if inside else 0.0,
            })
    return out


def write_csv(path: str, rows) -> None:
    with open(path, "w", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=list(rows[0]))
        w.writeheader()
        w.writerows(rows)


def report(models, acc, summary, bias, ladder=()) -> None:
    short = {m: m.replace("Qwen_Qwen", "").replace("-Instruct", "") for m in models}
    idx = {(r["model"], r["task"]): r for r in summary}

    print("=" * 96)
    print("ACCURACY  (mean over the six baselines; floor = what the split gives "
          "away for free)")
    print("=" * 96)
    print(f'{"model":22s}' + "".join(f"{t.replace('youtube_','yt_'):>16s}" for t in TASKS))
    print(f'{"floor":22s}' + "".join(f"{FLOORS[t]:>16.3f}" for t in TASKS))
    print("-" * 96)
    for m in models:
        print(f"{short[m]:22s}" + "".join(
            f'{idx[(m,t)]["acc_mean"]:>16.3f}' for t in TASKS))
    print()
    print("HEADROOM CAPTURED  (acc - floor) / (1 - floor);  negative = below floor")
    print("-" * 96)
    for m in models:
        cells = []
        for t in TASKS:
            v = idx[(m, t)]["headroom_captured_mean"]
            cells.append(f"{v:>+15.3f} " if v < 0 else f"{v:>16.3f}")
        print(f"{short[m]:22s}" + "".join(cells))
    print()
    print("BASELINE SENSITIVITY  range over 0.00-0.25  /  slope per unit baseline")
    print("-" * 96)
    for m in models:
        print(f"{short[m]:22s}" + "".join(
            f'{idx[(m,t)]["acc_range"]:>7.3f}/{idx[(m,t)]["slope_per_baseline"]:>+8.3f}'
            for t in TASKS))
    print()
    print("ANSWER COLLAPSE  modal predicted label, its share, and what always "
          "answering it would score (baseline 0.00)")
    print("-" * 96)
    bidx = {(r["model"], r["task"], r["baseline"]): r for r in bias}
    for m in models:
        cells = []
        for t in TASKS:
            r = bidx[(m, t, "0.00")]
            cells.append(f'{r["modal_prediction"]:>7s}{r["modal_share"]:>6.2f}'
                         f'{r["always_modal_accuracy"]:>7.3f}')
        print(f"{short[m]:22s}" + "".join(cells))
    print("  (reference labels are near-uniform, so 'always modal' sits at chance)")

    if ladder:
        print()
        print("LEVEL 2 REGRESSION GUARD -- can a text-only majority vote still help?")
        print("-" * 96)
        lidx = {(r["model"], r["baseline"]): r for r in ladder}
        for m in models:
            r = lidx[(m, "0.00")]
            print(f'{short[m]:22s} answer inside the vote survivors '
                  f'{r["answer_in_survivor_set"]:.3f}   accuracy {r["accuracy"]:.3f}')
        print("  distractors are random, so the vote rarely decides anything; a")
        print("  survivor rate near 1.0 means it eliminated nothing.")


if __name__ == "__main__":
    raise SystemExit(main())

	#!/usr/bin/env python3
	"""Compare every model across the four evaluation splits.

	python analysis/dataset_level_analysis/analyze_dataset_level.py

	Reads result/<model>/<task>/baseline_<b>/{metrics.json,predictions.jsonl} and
	writes the CSVs next to this script.

	Raw accuracy is not comparable across the four splits: VSR is a two-way
	true/false question while the three Youtube levels are four-way, and level 2
	carries a structural floor of its own (see FLOORS). Everything is therefore
	also reported as headroom captured, (acc - floor) / (1 - floor): the share of
	the distance between guessing and perfect that the model actually covers. It
	is negative when a model does worse than the floor, which happens here.
	"""
	from __future__ import annotations

	import argparse
	import collections
	import csv
	import json
	import math
	import os
	import statistics

	BASELINES = ["0.00", "0.05", "0.10", "0.15", "0.20", "0.25"]
	TASKS = ["vsr", "youtube_level1", "youtube_level2", "youtube_level3"]

	# What a model scores on each split without looking at the image: pure guessing
	# among the choices. Level 2's three distractors are drawn uniformly from the
	# 23 wrong orderings, so the four options are exchangeable and every text-only
	# heuristic sits at chance -- measured on the split: prefix vote 0.252, medoid
	# 0.256, outlier 0.254, positional majority 0.245. Earlier structured variants
	# did NOT have this property; see disparity_qa/level2_choices.py.
	FLOORS = {"vsr": 0.500, "youtube_level1": 0.250,
	"youtube_level2": 0.250, "youtube_level3": 0.250}
	CHANCE = dict(FLOORS)


	def load_metrics(root: str, model: str, task: str, baseline: str) -> dict:
	with open(os.path.join(root, "result", model, task,
	f"baseline_{baseline}", "metrics.json")) as fh:
	return json.load(fh)


	def load_predictions(root: str, model: str, task: str, baseline: str):
	path = os.path.join(root, "result", model, task,
	f"baseline_{baseline}", "predictions.jsonl")
	with open(path) as fh:
	for line in fh:
	line = line.strip()
	if line:
	yield json.loads(line)


	def entropy(counter: collections.Counter) -> float:
	n = sum(counter.values())
	if not n:
	return 0.0
	return -sum((v / n) * math.log2(v / n) for v in counter.values() if v)


	def slope(xs, ys) -> float:
	"""Least-squares slope of accuracy against baseline (per unit of baseline)."""
	mx, my = statistics.mean(xs), statistics.mean(ys)
	den = sum((x - mx) ** 2 for x in xs)
	return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / den if den else 0.0


	def main() -> int:
	here = os.path.dirname(os.path.abspath(__file__))
	p = argparse.ArgumentParser()
	p.add_argument("--root", default=os.path.abspath(os.path.join(here, "..", "..")))
	p.add_argument("--out_dir", default=here)
	args = p.parse_args()

	models = sorted(m for m in os.listdir(os.path.join(args.root, "result"))
	if not m.startswith("_")
	and os.path.isdir(os.path.join(args.root, "result", m)))

	acc = {} # (model, task, baseline) -> metrics dict
	for m in models:
	for t in TASKS:
	for b in BASELINES:
	acc[(m, t, b)] = load_metrics(args.root, m, t, b)

	# ---------------------------------------------------------- per baseline
	rows = []
	for m in models:
	for t in TASKS:
	for b in BASELINES:
	d = acc[(m, t, b)]
	rows.append({
	"model": m, "task": t, "baseline": b,
	"n": d["num_samples"], "accuracy": round(d["accuracy"], 4),
	"num_correct": d["num_correct"],
	"unparsed_rate": round(d["unparsed_rate"], 4),
	"scoring": d["scoring"],
	"chance": CHANCE[t], "floor": FLOORS[t],
	"headroom_captured": round(
	(d["accuracy"] - FLOORS[t]) / (1 - FLOORS[t]), 4),
	})
	write_csv(os.path.join(args.out_dir, "accuracy_by_baseline.csv"), rows)

	# ------------------------------------------------------- per model/task
	summary = []
	for m in models:
	for t in TASKS:
	a = [acc[(m, t, b)]["accuracy"] for b in BASELINES]
	xs = [float(b) for b in BASELINES]
	summary.append({
	"model": m, "task": t, "n": acc[(m, t, "0.00")]["num_samples"],
	"scoring": acc[(m, t, "0.00")]["scoring"],
	"chance": CHANCE[t], "floor": FLOORS[t],
	"acc_b0": round(a[0], 4), "acc_mean": round(statistics.mean(a), 4),
	"acc_min": round(min(a), 4), "acc_max": round(max(a), 4),
	"acc_range": round(max(a) - min(a), 4),
	"acc_std": round(statistics.pstdev(a), 4),
	"slope_per_baseline": round(slope(xs, a), 4),
	"delta_b025_minus_b0": round(a[-1] - a[0], 4),
	"headroom_captured_mean": round(
	(statistics.mean(a) - FLOORS[t]) / (1 - FLOORS[t]), 4),
	"above_floor": statistics.mean(a) > FLOORS[t],
	})
	write_csv(os.path.join(args.out_dir, "summary_by_model_task.csv"), summary)

	# --------------------------------------------------------------- biases
	bias = []
	for m in models:
	for t in TASKS:
	for b in BASELINES:
	pred = collections.Counter()
	ref = collections.Counter()
	n = 0
	for r in load_predictions(args.root, m, t, b):
	pred[r["prediction_norm"]] += 1
	ref[r["reference_norm"]] += 1
	n += 1
	top_label, top_n = pred.most_common(1)[0]
	# what a model would score by always naming its own modal answer
	majority_acc = ref[top_label] / n if n else 0.0
	bias.append({
	"model": m, "task": t, "baseline": b, "n": n,
	"n_distinct_predictions": len(pred),
	"modal_prediction": top_label,
	"modal_share": round(top_n / n, 4),
	"prediction_entropy_bits": round(entropy(pred), 4),
	"reference_entropy_bits": round(entropy(ref), 4),
	"always_modal_accuracy": round(majority_acc, 4),
	"accuracy": round(acc[(m, t, b)]["accuracy"], 4),
	"prediction_counts": json.dumps(dict(sorted(pred.items()))),
	})
	write_csv(os.path.join(args.out_dir, "prediction_bias.csv"), bias)

	ladder = level2_ladder(args.root, models)
	if ladder:
	write_csv(os.path.join(args.out_dir, "level2_structure_use.csv"), ladder)

	report(models, acc, summary, bias, ladder)
	print(f"\nwrote {3 + bool(ladder)} CSVs to {args.out_dir}")
	return 0


	# the split lives in the repo, one level above this experiment directory
	LEVEL2_TEST = os.path.join("..", "02_data", "Youtube_self_depth_QA",
	"level2", "test.jsonl")


	def _prefix_survivors(options):
	"""The options a majority vote on successive positions cannot rule out.

	A regression guard. With random distractors this should leave most items
	undecided and score at chance; if it ever climbs toward 0.5 again, the
	option construction has reacquired a structural leak.
	"""
	alive = list(options)
	seqs = {lab: txt.split(" > ") for lab, txt in options.items()}
	for k in range(4):
	if len(alive) <= 1:
	break
	counts = collections.Counter(seqs[a][k] for a in alive)
	top = counts.most_common(1)[0][1]
	winners = [l for l, v in counts.items() if v == top]
	if len(winners) == 1:
	alive = [a for a in alive if seqs[a][k] == winners[0]]
	return alive


	def level2_ladder(root: str, models):
	"""How often each model's answer lands inside the structure-only survivor set."""
	path = os.path.abspath(os.path.join(root, LEVEL2_TEST))
	if not os.path.exists(path):
	print(f"[warn] level-2 split not found at {path}; skipping ladder analysis")
	return []
	test = {}
	with open(path) as fh:
	for line in fh:
	rec = json.loads(line)
	if rec.get("options"):
	test[rec["sample_id"]] = rec["options"]
	if not test:
	return []
	out = []
	for m in models:
	for b in BASELINES:
	inside = hit = n = 0
	for r in load_predictions(root, m, "youtube_level2", b):
	opts = test.get(r["sample_id"])
	if not opts:
	continue
	alive = _prefix_survivors(opts)
	n += 1
	inside += int(r["prediction_norm"] in alive)
	hit += int(r["correct"])
	out.append({
	"model": m, "baseline": b, "n": n,
	"answer_in_survivor_set": round(inside / n, 4) if n else 0.0,
	"accuracy": round(hit / n, 4) if n else 0.0,
	# of the answers that did land in the survivor set, how many were
	# right: this is the coin flip the structure cannot decide
	"accuracy_within_survivors": round(hit / inside, 4) if inside else 0.0,
	})
	return out


	def write_csv(path: str, rows) -> None:
	with open(path, "w", newline="") as fh:
	w = csv.DictWriter(fh, fieldnames=list(rows[0]))
	w.writeheader()
	w.writerows(rows)


	def report(models, acc, summary, bias, ladder=()) -> None:
	short = {m: m.replace("Qwen_Qwen", "").replace("-Instruct", "") for m in models}
	idx = {(r["model"], r["task"]): r for r in summary}

	print("=" * 96)
	print("ACCURACY (mean over the six baselines; floor = what the split gives "
	"away for free)")
	print("=" * 96)
	print(f'{"model":22s}' + "".join(f"{t.replace('youtube_','yt_'):>16s}" for t in TASKS))
	print(f'{"floor":22s}' + "".join(f"{FLOORS[t]:>16.3f}" for t in TASKS))
	print("-" * 96)
	for m in models:
	print(f"{short[m]:22s}" + "".join(
	f'{idx[(m,t)]["acc_mean"]:>16.3f}' for t in TASKS))
	print()
	print("HEADROOM CAPTURED (acc - floor) / (1 - floor); negative = below floor")
	print("-" * 96)
	for m in models:
	cells = []
	for t in TASKS:
	v = idx[(m, t)]["headroom_captured_mean"]
	cells.append(f"{v:>+15.3f} " if v < 0 else f"{v:>16.3f}")
	print(f"{short[m]:22s}" + "".join(cells))
	print()
	print("BASELINE SENSITIVITY range over 0.00-0.25 / slope per unit baseline")
	print("-" * 96)
	for m in models:
	print(f"{short[m]:22s}" + "".join(
	f'{idx[(m,t)]["acc_range"]:>7.3f}/{idx[(m,t)]["slope_per_baseline"]:>+8.3f}'
	for t in TASKS))
	print()
	print("ANSWER COLLAPSE modal predicted label, its share, and what always "
	"answering it would score (baseline 0.00)")
	print("-" * 96)
	bidx = {(r["model"], r["task"], r["baseline"]): r for r in bias}
	for m in models:
	cells = []
	for t in TASKS:
	r = bidx[(m, t, "0.00")]
	cells.append(f'{r["modal_prediction"]:>7s}{r["modal_share"]:>6.2f}'
	f'{r["always_modal_accuracy"]:>7.3f}')
	print(f"{short[m]:22s}" + "".join(cells))
	print(" (reference labels are near-uniform, so 'always modal' sits at chance)")

	if ladder:
	print()
	print("LEVEL 2 REGRESSION GUARD -- can a text-only majority vote still help?")
	print("-" * 96)
	lidx = {(r["model"], r["baseline"]): r for r in ladder}
	for m in models:
	r = lidx[(m, "0.00")]
	print(f'{short[m]:22s} answer inside the vote survivors '
	f'{r["answer_in_survivor_set"]:.3f} accuracy {r["accuracy"]:.3f}')
	print(" distractors are random, so the vote rarely decides anything; a")
	print(" survivor rate near 1.0 means it eliminated nothing.")


	if __name__ == "__main__":
	raise SystemExit(main())