exp01_sorting.py
#!/usr/bin/env python3
"""Experiment 0.1: sorting algorithms as minimal agents, with controls.
Reimplements Zhang, Goldstein, Levin 2024 (arXiv 2401.05375; code
github.com/Zhangtaining/cell_research) plus the controls in plan.md:
deterministic schedulers, a metric-substitution set, per-seed intervals.
Cell rules follow the authors' code, not the paper's prose, where they differ:
bubble each turn picks left or right at random; swaps if out of order
insertion moves left if smaller than its left neighbour and the prefix to its
left, ignoring frozen cells, is sorted
selection holds a pointer starting at position 0; compares itself with the
occupant; pointer advances if it is not smaller, else it swaps in
Frozen cells in their code never initiate a move but can be swapped by an
active cell ("movable" here). "immovable" (no swap touching it) is the paper's
second kind and is our extension. Classical variants with frozen cells are not
in their repo; ours block the initiating element (movable) or any swap
(immovable).
Arrays are permutations of 0..n-1. A step is one successful swap; metrics are
sampled at swaps, since nothing else changes them.
Metrics as "higher is better" fractions:
adj 1 - adjacent descents / (n-1) the paper's Sortedness
kendall 1 - total inversions / C(n,2) the potential function
own bubble only: fraction of cells with both neighbours in order
DG (ours) under a metric: gains made while below the running maximum; a
monotone metric gives 0. DG (paper): their avg_wandering_range on the
adjacent-descent series, ported verbatim from analysis/.
Usage: exp01_sorting.py [--n 64] [--seeds 100] [--out results/exp01]
exp01_sorting.py --selftest
"""
import argparse
import csv
import inspect
import os
import random
import sys
import zlib
ALGOS = ("bubble", "insertion", "selection")
FROZEN_KINDS = ("movable", "immovable")
class Array:
"""Permutation with incrementally maintained inversion counts."""
def __init__(self, vals, frozen=None):
self.vals = list(vals)
self.n = len(vals)
self.frozen = dict(frozen or {}) # value -> kind
self.bad = {k for k in range(1, self.n) if self.vals[k] < self.vals[k - 1]}
self.inv = sum(
1 for i in range(self.n) for j in range(i + 1, self.n) if self.vals[i] > self.vals[j]
)
self.swaps = 0
self.sat = sum(1 for i in range(self.n) if self.calm(i))
def adj(self):
return 1.0 - len(self.bad) / (self.n - 1)
def kendall(self):
return 1.0 - self.inv / (self.n * (self.n - 1) / 2)
def own(self):
return self.sat / self.n
def calm(self, i):
"""Bubble cell at i wants no move: both neighbours in order."""
v = self.vals[i]
return not (i > 0 and v < self.vals[i - 1]) and not (i < self.n - 1 and v > self.vals[i + 1])
def first_unsorted(self, skip_frozen=False):
if not skip_frozen:
return min(self.bad) if self.bad else self.n
prev = -1
for i in range(self.n):
v = self.vals[i]
if v in self.frozen:
prev = -1 # authors' code resets the check after a frozen cell
continue
if v < prev:
return i
prev = v
return self.n
def swap(self, i, j, active):
"""Swap positions i, j unless a frozen cell forbids it. active is the
initiating cell's value. Returns True on success."""
if i > j:
i, j = j, i
a, b = self.vals[i], self.vals[j]
for v in (a, b):
kind = self.frozen.get(v)
if kind == "immovable" or (kind == "movable" and v == active):
return False
near = [k for k in range(i - 1, j + 2) if 0 <= k < self.n]
self.sat -= sum(1 for k in near if self.calm(k))
d = 1 if a < b else -1
for k in range(i + 1, j):
c = self.vals[k]
d += 1 if c < b else -1 # b now precedes c
d += 1 if c > a else -1 # c now precedes a
self.inv += d
self.vals[i], self.vals[j] = b, a
for k in (i, i + 1, j, j + 1):
if 1 <= k < self.n:
if self.vals[k] < self.vals[k - 1]:
self.bad.add(k)
else:
self.bad.discard(k)
self.sat += sum(1 for k in near if self.calm(k))
self.swaps += 1
return True
# --- cell rules: return the position the cell at i wants to swap with, or None ---
def rule_bubble(arr, i, ctx):
v = arr.vals[i]
if ctx.rng.random() < 0.5:
return i + 1 if i < arr.n - 1 and v > arr.vals[i + 1] else None
return i - 1 if i > 0 and v < arr.vals[i - 1] else None
def rule_insertion(arr, i, ctx):
v = arr.vals[i]
if i > 0 and v < arr.vals[i - 1] and ctx.fu >= i:
return i - 1
return None
def rule_selection(arr, i, ctx):
v = arr.vals[i]
p = ctx.ptr[v]
if p == i or p >= arr.n:
return None
occupant = arr.vals[p]
if occupant in arr.frozen:
ctx.ptr[v] = p + 1
ctx.active = True
return p if v < occupant else None
if v >= occupant:
ctx.ptr[v] = p + 1
ctx.active = True
return None
return p
RULES = {"bubble": rule_bubble, "insertion": rule_insertion, "selection": rule_selection}
class Ctx:
def __init__(self, arr, rng):
self.rng = rng
self.ptr = dict.fromkeys(range(arr.n), 0)
self.fu = 0
self.active = False # a pointer advanced this round
def run_cellview(arr, algotype, sched, rng, max_rounds=None, p_move=None):
"""Yield after every successful swap. algotype: value -> algo name.
p_move: value -> probability of attempting a move on a turn (default 1)."""
max_rounds = max_rounds or arr.n * arr.n # insertion makes ~1 swap per round
ctx = Ctx(arr, rng)
order = list(range(arr.n))
idle = 0
for _ in range(max_rounds):
if sched == "random":
rng.shuffle(order)
ctx.fu = arr.first_unsorted(skip_frozen=True)
ctx.active = False
moved = False
for i in order:
v = arr.vals[i]
if v in arr.frozen:
continue # frozen cells never initiate
if p_move is not None and rng.random() >= p_move.get(v, 1.0):
continue
j = RULES[algotype[v]](arr, i, ctx)
if j is not None and arr.swap(i, j, active=v):
moved = True
ctx.fu = arr.first_unsorted(skip_frozen=True)
yield
# bubble's random side can idle a round while a cell still wants to
# move; stop only when no bubble cell wants to and nothing advanced
idle = 0 if moved or ctx.active else idle + 1
if idle >= 3 and all(arr.calm(i) for i in order if algotype[arr.vals[i]] == "bubble"):
return
def run_classical(arr, algo):
n = arr.n
if algo == "bubble":
for end in range(n - 1, 0, -1):
moved = False
for i in range(end):
if arr.vals[i] > arr.vals[i + 1] and arr.swap(i, i + 1, active=arr.vals[i]):
moved = True
yield
if not moved:
return
elif algo == "insertion":
for i in range(1, n):
j = i
while j > 0 and arr.vals[j] < arr.vals[j - 1]:
if not arr.swap(j - 1, j, active=arr.vals[j]):
break
yield
j -= 1
elif algo == "selection":
for i in range(n - 1):
m = min(range(i, n), key=lambda k: arr.vals[k])
if m != i and arr.swap(i, m, active=arr.vals[m]):
yield
# --- measurement ---
def dg(series):
total, best = 0.0, series[0]
for prev, cur in zip(series, series[1:]):
if cur > prev and prev < best:
total += cur - prev
best = max(best, cur)
return total
def dg_paper(descents):
"""Authors' avg_wandering_range on the adjacent-descent count series."""
arr = []
for x in descents:
if not arr or arr[-1] != x:
arr.append(x)
temp = [arr[i] - arr[i - 1] for i in range(1, len(arr))]
if not temp:
return 0.0
runs, x = [], temp[0]
for i in range(1, len(temp)):
if temp[i - 1] * temp[i] > 0:
x += temp[i]
else:
runs.append(x)
x = temp[i]
runs.append(x)
i = next((k for k, d in enumerate(runs) if d > 0), len(runs))
res, cnt = 0.0, 0
while i < len(runs) - 1:
if runs[i + 1] > 0:
break
res += -(runs[i + 1] + runs[i]) / runs[i]
cnt += 1
i += 2
return res / cnt if cnt else 0.0
def trace(arr, runner, with_own):
adj, ken, own, desc = [arr.adj()], [arr.kendall()], [arr.own()], [len(arr.bad)]
for _ in runner:
adj.append(arr.adj())
ken.append(arr.kendall())
own.append(arr.own())
desc.append(len(arr.bad))
return {
"swaps": arr.swaps,
"final_adj_inv": len(arr.bad),
"final_inv": arr.inv,
"dg_adj": dg(adj),
"dg_kendall": dg(ken),
"dg_own": dg(own) if with_own else float("nan"),
"dg_paper": dg_paper(desc),
}
def make_array(n, rng, n_frozen, kind):
vals = list(range(n))
rng.shuffle(vals)
frozen = {v: kind for v in rng.sample(range(n), n_frozen)}
return Array(vals, frozen)
def mean_ci(xs, rng, reps=1000):
m = sum(xs) / len(xs)
boots = sorted(sum(rng.choice(xs) for _ in xs) / len(xs) for _ in range(reps))
return m, boots[int(0.025 * reps)], boots[int(0.975 * reps)]
def effort_bits(fn):
return 8 * len(zlib.compress(inspect.getsource(fn).encode()))
KEYS = ("swaps", "final_adj_inv", "final_inv", "dg_adj", "dg_kendall", "dg_own", "dg_paper")
def summarise(res, rng):
row = {}
for key in KEYS:
xs = [r[key] for r in res]
if xs[0] != xs[0]: # nan
row[key] = row[key + "_lo"] = row[key + "_hi"] = ""
continue
row[key], row[key + "_lo"], row[key + "_hi"] = mean_ci(xs, rng)
return row
def run_grid(n, seeds, out_dir):
rows = []
for algo in ALGOS:
for variant in ("classical", "cellview"):
scheds = ("roundrobin", "random") if variant == "cellview" else ("none",)
for sched in scheds:
for kind in FROZEN_KINDS:
for nf in (0, 1, 2, 3):
if nf == 0 and kind == "immovable":
continue
res = []
for s in range(seeds):
rng = random.Random(1000 * s + 7)
arr = make_array(n, rng, nf, kind)
if variant == "classical":
r = trace(arr, run_classical(arr, algo), False)
else:
algotype = dict.fromkeys(range(n), algo)
r = trace(arr, run_cellview(arr, algotype, sched, rng), algo == "bubble")
res.append(r)
row = {
"algo": algo,
"variant": variant,
"sched": sched,
"frozen_kind": kind if nf else "none",
"n_frozen": nf,
"n": n,
"seeds": seeds,
"effort_bits": effort_bits(RULES[algo] if variant == "cellview" else run_classical),
}
row.update(summarise(res, random.Random(1)))
rows.append(row)
own = "" if row["dg_own"] == "" else f"{row['dg_own']:.3f}"
print(
f"{algo:9} {variant:9} {sched:10} {row['frozen_kind']:9} nf={nf} "
f"swaps={row['swaps']:7.1f} inv={row['final_inv']:6.1f} "
f"dg_adj={row['dg_adj']:.3f} dg_ken={row['dg_kendall']:.3f} "
f"dg_own={own:5} dg_paper={row['dg_paper']:.3f}",
file=sys.stderr,
)
os.makedirs(out_dir, exist_ok=True)
with open(os.path.join(out_dir, "grid.csv"), "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
w.writeheader()
w.writerows(rows)
return rows
def aggregation(arr, label):
"""Fraction of cells whose neighbours all share their label."""
n, v = arr.n, arr.vals
same = 0
for i in range(n):
left = i == 0 or label[v[i - 1]] == label[v[i]]
right = i == n - 1 or label[v[i + 1]] == label[v[i]]
if left and right:
same += 1
return same / n
def run_chimera(n, seeds, out_dir):
"""Chimeric arrays (half each of two algotypes) vs a label-only control that
runs one algorithm everywhere. peak is biased by run length; mean is not."""
pairs = [("bubble", "insertion"), ("bubble", "selection"), ("insertion", "selection")]
rows = []
for a, b in pairs:
for control in (None, a, b):
peaks, means, finals, invs = [], [], [], []
for s in range(seeds):
rng = random.Random(1000 * s + 7)
arr = 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), control) if control else label
series = [aggregation(arr, label)]
for _ in run_cellview(arr, algotype, "roundrobin", rng):
series.append(aggregation(arr, label))
peaks.append(max(series))
means.append(sum(series) / len(series))
finals.append(series[-1])
invs.append(arr.inv)
rng = random.Random(1)
row = {"pair": f"{a}+{b}", "control": control or "chimera", "n": n, "seeds": seeds}
for key, xs in (("peak", peaks), ("mean", means), ("final", finals), ("final_inv", invs)):
row[key], row[key + "_lo"], row[key + "_hi"] = mean_ci(xs, rng)
rows.append(row)
print(
f"chimera {a}+{b:9} {row['control']:9} peak={row['peak']:.3f} "
f"mean={row['mean']:.3f} [{row['mean_lo']:.3f},{row['mean_hi']:.3f}] "
f"final={row['final']:.3f} inv={row['final_inv']:.1f}",
file=sys.stderr,
)
with open(os.path.join(out_dir, "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(3)
for algo in ALGOS:
for variant in ("classical", "cellview"):
arr = make_array(40, rng, 0, "movable")
algotype = dict.fromkeys(range(40), algo)
runner = run_classical(arr, algo) if variant == "classical" else run_cellview(arr, algotype, "random", rng)
ken = [arr.kendall()]
for _ in runner:
recount = sum(1 for i in range(40) for j in range(i + 1, 40) if arr.vals[i] > arr.vals[j])
assert arr.inv == recount, (algo, variant, arr.inv, recount)
assert arr.bad == {k for k in range(1, 40) if arr.vals[k] < arr.vals[k - 1]}
assert arr.sat == sum(1 for i in range(40) if arr.calm(i))
ken.append(arr.kendall())
assert arr.vals == list(range(40)), (algo, variant, arr.vals)
if algo != "selection":
assert dg(ken) == 0.0, (algo, variant, dg(ken))
arr = Array([3, 1, 0, 2], {0: "immovable"})
for _ in run_cellview(arr, dict.fromkeys(range(4), "bubble"), "roundrobin", rng):
pass
assert arr.vals[2] == 0 and arr.vals != [0, 1, 2, 3]
arr = Array([3, 1, 0, 2], {3: "movable"})
for _ in run_cellview(arr, dict.fromkeys(range(4), "bubble"), "roundrobin", rng):
pass
assert arr.vals == [0, 1, 2, 3], arr.vals
assert dg([1, 0.5, 0.8, 1.0, 0.9, 1.0]) == 0.5 + 0.1
# paper formula: descents 3,4,4,2,1 -> runs [+1,-3]; (3-1)/1 = 2
assert dg_paper([3, 4, 4, 2, 1]) == 2.0
assert dg_paper([3, 2, 1, 0]) == 0.0
print("selftest ok")
def main():
p = argparse.ArgumentParser()
p.add_argument("--n", type=int, default=64)
p.add_argument("--seeds", type=int, default=100)
p.add_argument("--out", default=os.path.join(os.path.dirname(__file__), "results", "exp01"))
p.add_argument("--selftest", action="store_true")
a = p.parse_args()
if a.selftest:
selftest()
return
run_grid(a.n, a.seeds, a.out)
run_chimera(a.n, a.seeds, a.out)
if __name__ == "__main__":
main()