exp04_rate_chimera.py

#!/usr/bin/env python3
"""Experiment 0.1b: rate-matched chimera.

All cells run the bubble rule; label-B cells attempt a move with probability p
per turn. If algotype clustering in 0.1 is a rate effect, clustering should be
absent at p = 1 and rise as p falls. Predictions in plan.md.

Usage: exp04_rate_chimera.py [--n 100] [--seeds 100] [--out results/exp04]
       exp04_rate_chimera.py --selftest
"""

import argparse
import csv
import os
import random
import sys

import exp01_sorting as s

PS = (1.0, 0.75, 0.5, 0.25, 0.1)


def run(n, seeds, out_dir):
    rows = []
    for p in PS:
        means, peaks, finals = [], [], []
        for seed in range(seeds):
            rng = random.Random(1000 * seed + 7)
            arr = s.make_array(n, rng, 0, "movable")
            labels = ["A"] * (n // 2) + ["B"] * (n - n // 2)
            rng.shuffle(labels)
            label = dict(enumerate(labels))
            algotype = dict.fromkeys(range(n), "bubble")
            p_move = {v: (p if label[v] == "B" else 1.0) for v in range(n)}
            series = [s.aggregation(arr, label)]
            for _ in s.run_cellview(arr, algotype, "roundrobin", rng, p_move=p_move):
                series.append(s.aggregation(arr, label))
            means.append(sum(series) / len(series))
            peaks.append(max(series))
            finals.append(series[-1])
        rng = random.Random(1)
        row = {"p": p, "n": n, "seeds": seeds}
        for key, xs in (("mean", means), ("peak", peaks), ("final", finals)):
            row[key], row[key + "_lo"], row[key + "_hi"] = s.mean_ci(xs, rng)
        rows.append(row)
        print(f"p={p:4} mean={row['mean']:.3f} [{row['mean_lo']:.3f},{row['mean_hi']:.3f}] "
              f"peak={row['peak']:.3f} final={row['final']:.3f}", file=sys.stderr)
    os.makedirs(out_dir, exist_ok=True)
    with open(os.path.join(out_dir, "rate_chimera.csv"), "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)
    return rows


def selftest():
    rng = random.Random(0)
    arr = s.make_array(30, rng, 0, "movable")
    algotype = dict.fromkeys(range(30), "bubble")
    p_move = {v: 0.3 for v in range(30)}
    for _ in s.run_cellview(arr, algotype, "roundrobin", rng, p_move=p_move):
        pass
    assert arr.vals == list(range(30)), arr.vals
    print("selftest ok")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--n", type=int, default=100)
    ap.add_argument("--seeds", type=int, default=100)
    ap.add_argument("--out", default=os.path.join(os.path.dirname(__file__), "results", "exp04"))
    ap.add_argument("--selftest", action="store_true")
    a = ap.parse_args()
    if a.selftest:
        selftest()
        return
    run(a.n, a.seeds, a.out)


if __name__ == "__main__":
    main()