estate-rule sim harness: stdlib core, runner, tests

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Heimdall 2026-11-05 03:39:53 +00:00
commit dda8ee7c58
3 changed files with 844 additions and 0 deletions

View file

@ -0,0 +1,371 @@
"""Estate-rule simulation core. Pure stdlib. Spec: SPEC.md in this directory.
One replicate = one population (delegation network, strategic flags, faction)
plus a sequence of motions. Every random draw comes from two Random objects
seeded from integers, so a (Config, rep) pair is a pure function.
"""
import math
import random
from dataclasses import dataclass
ESTATES = ("Minds", "Members", "Users", "Patrons")
ROLL = {"Minds": 8, "Members": 15, "Users": 300, "Patrons": 4}
BASE = {
"Minds": dict(a=1.0, b=.15, s=.9, sigma=.7, turnout=1.0, kappa=0.0),
"Members": dict(a=.8, b=.4, s=.5, sigma=1.0, turnout=.8, kappa=1.0),
"Users": dict(a=.4, b=.9, s=.5, sigma=1.2, turnout=.10, kappa=1.0),
"Patrons": dict(a=.7, b=.5, s=.5, sigma=1.0, turnout=.8, kappa=1.0),
}
DELEG_SHARE = {"Minds": 0.0, "Members": .2, "Users": .5, "Patrons": .2}
WEIGHTS = {"equal": (25, 25, 25, 25), "minds-heavy": (40, 30, 20, 10)}
CATS = (.15, .30, .40, .50, .66)
RULES = ("DIRECT", "HEAD", "WTA", "UNI")
ABSTAIN_BAND = .1
STRONG = .5 # |u| above which a final holder's opposing ballot is a strike
STRIKES = 2 # strikes that revoke a delegation for the rest of the replicate
SCOPE_P = .6
DELEGATE_FRAC = .1
POPULIST_DELEGATE_P = .25
EPS = 1e-9
@dataclass(frozen=True)
class Config:
weights: str = "equal"
delegation: bool = False
alpha: float = 1.0 # Zipf exponent of delegate choice (concentration)
beta: float = 0.0 # extra attraction of populist delegates
cap: float = None # per-final-holder cap as a fraction of cap_basis; None = no cap
cap_basis: str = "cast" # "cast": estate weight resolved before capping; "roll": estate roll size
depth: int = 3 # max delegation edges a vote may traverse; None = unlimited
dturn: float = .95 # delegates' own turnout; a delegate who stays home passes weight onward
onward: float = .3 # probability an active delegate itself delegates onward
rho: float = 0.0 # strategic fraction
faction: int = 0 # coordinated Users faction size
motions: int = 400
overrides: tuple = () # ((estate, key, value), ...)
# ---------------------------------------------------------------- rules
def passes(rule, F, A, W, q):
"""Does the motion pass? F, A: per-estate resolved For/Against weight; W: estate weights."""
if rule == "HEAD":
f, a = sum(F), sum(A)
return f + a > 0 and f >= q * (f + a) - EPS
fo = tot = 0.0
for e in range(len(F)):
n = F[e] + A[e]
if n <= 0:
continue
if rule == "DIRECT":
fo += W[e] * F[e] / n
elif rule == "WTA":
if F[e] == A[e]:
continue # tied estate casts nothing and drops out
if F[e] > A[e]:
fo += W[e]
elif rule == "UNI":
if F[e] >= q * n - EPS:
fo += W[e]
else:
raise ValueError(rule)
tot += W[e]
return tot > 0 and fo >= q * tot - EPS
# ---------------------------------------------------------------- population
class Estate:
def __init__(self, name, cfg, rng):
n = ROLL[name]
p = dict(BASE[name])
for en, k, v in cfg.overrides:
if en == name:
p[k] = v
self.name, self.n, self.cfg = name, n, cfg
self.a = [p["a"]] * n
self.b = [p["b"]] * n
self.s = p["s"]
self.sig = [p["sigma"]] * n
self.turn = [p["turnout"]] * n
self.kap = [p["kappa"]] * n
self.target = [None] * n
self.scope = [None] * n
self.strikes = [0] * n
self.is_delegate = [False] * n
self.populist = [False] * n
self.faction = [False] * n
self.strategic = [rng.random() < cfg.rho for _ in range(n)]
ids = list(range(n))
rng.shuffle(ids)
rest = ids
share = DELEG_SHARE[name]
if cfg.delegation and share > 0:
nd = max(1, round(DELEGATE_FRAC * n))
delegates, rest = ids[:nd], ids[nd:]
for d in delegates:
self.is_delegate[d] = True
self.populist[d] = rng.random() < POPULIST_DELEGATE_P
if self.populist[d]:
self.a[d], self.b[d], self.sig[d] = .2, 1.2, .7
else:
self.a[d], self.b[d], self.sig[d] = .7, .5, .7
self.turn[d], self.kap[d] = cfg.dturn, 0.0
wts = [(k + 1) ** -cfg.alpha * ((1 + cfg.beta) if self.populist[d] else 1.0)
for k, d in enumerate(delegates)]
nde = min(len(rest), round(share * n))
delegators, rest = rest[:nde], rest[nde:]
for i in delegators:
self.target[i] = rng.choices(delegates, wts)[0]
self.scope[i] = self._scope(rng)
if len(delegates) > 1:
for k, d in enumerate(delegates):
if rng.random() < cfg.onward:
others = [(x, w) for x, w in zip(delegates, wts) if x != d]
self.target[d] = rng.choices([x for x, _ in others], [w for _, w in others])[0]
self.scope[d] = self._scope(rng)
if name == "Users":
for i in rest[:cfg.faction]:
self.faction[i] = True
self.ubar = [.8 * math.sqrt(self.a[i] ** 2 + self.b[i] ** 2 + self.s ** 2 + self.sig[i] ** 2)
for i in range(n)]
@staticmethod
def _scope(rng):
sc = frozenset(c for c in range(len(CATS)) if rng.random() < SCOPE_P)
return sc or frozenset((rng.randrange(len(CATS)),))
def chain(self, cat, voting):
"""Final holder of every non-voting node's weight this motion.
Follows delegation edges until a node that votes directly. The chain is
void (vote uncast) when any link's scope excludes cat (the usable scope
is the overlap of all links), it loops, it exceeds cfg.depth edges, or it
ends at a node with no edge. Returns (dest, uncast): dest maps node ->
final holder, uncast counts non-voting nodes that hold an edge but are void.
"""
D = self.cfg.depth
res = {} # node -> (holder, edges) or None for void
for i in range(self.n):
if voting[i] or self.target[i] is None or i in res:
continue
path, seen, cur = [], set(), i
out = None
while True:
if voting[cur]:
out = (cur, 0)
break
if cur in res:
out = res[cur]
break
if cur in seen or self.target[cur] is None or cat not in self.scope[cur]:
break # loop, dead end or out of scope
seen.add(cur)
path.append(cur)
cur = self.target[cur]
for node in reversed(path): # void propagates upstream; distances add
out = (out[0], out[1] + 1) if out else None
res[node] = out
dest, uncast = {}, 0
for i, r in res.items():
if voting[i] or self.target[i] is None:
continue
if r is not None and (D is None or r[1] <= D):
dest[i] = r[0]
else:
uncast += 1
return dest, uncast
def resolve(self, cat, voting):
dest, uncast = self.chain(cat, voting)
w = {}
for i in range(self.n):
if voting[i]:
w[i] = w.get(i, 0.0) + 1.0
for i, h in dest.items():
w[h] = w.get(h, 0.0) + 1.0
if self.cfg.cap is not None:
base = self.n if self.cfg.cap_basis == "roll" else sum(w.values())
cv = max(1.0, self.cfg.cap * base)
for t in w:
if w[t] > cv:
w[t] = cv # excess is void, not redistributed
self.last_uncast = uncast
return w, dest
def motion(self, cat, m, d, rng, z):
"""Draw this estate's motion state; returns dict with resolved holders and joiners."""
n = self.n
u = [self.a[i] * m + self.b[i] * d + self.s * z + self.sig[i] * rng.gauss(0, 1) for i in range(n)]
un = [rng.random() for _ in range(n)]
agenda = d - m > 0
voting, sb = [False] * n, [0] * n
for i in range(n):
if self.kap[i]:
p = self.turn[i] * (1 + self.kap[i] * (abs(u[i]) - self.ubar[i]))
p = min(1.0, max(0.0, p))
else:
p = self.turn[i]
voting[i] = un[i] < p
sb[i] = 1 if u[i] > ABSTAIN_BAND else (-1 if u[i] < -ABSTAIN_BAND else 0)
if self.faction[i]:
voting[i] = True
sb[i] = 1 if agenda else -1
w, dest = self.resolve(cat, voting)
forced = False
for i, dd in dest.items():
if abs(u[i]) > STRONG and sb[dd] * u[i] < 0:
voting[i] = True
forced = True
self.strikes[i] += 1
if self.strikes[i] >= STRIKES:
self.target[i] = None # revoked for the rest of the replicate
if forced:
w, dest = self.resolve(cat, voting)
holders = [(t, sb[t], w[t]) for t in w]
joiners = [(i, sb[i]) for i in range(n)
if not voting[i] and i not in dest and sb[i] != 0]
return dict(u=u, holders=holders, joiners=joiners, weights=sorted(w.values()),
uncast=self.last_uncast, delegated=len(dest) + self.last_uncast)
def concentration(weights):
"""(top share, gini, kish n_eff, total) of resolved holder weights."""
tot = sum(weights)
if not weights or tot <= 0:
return (0.0, 0.0, 0.0, 0.0)
ws = sorted(weights)
n = len(ws)
cum = sum((k + 1) * x for k, x in enumerate(ws))
gini = (2 * cum) / (n * tot) - (n + 1) / n
neff = tot * tot / sum(x * x for x in ws)
return (ws[-1] / tot, gini, neff, tot)
# ---------------------------------------------------------------- one motion
def strategize(rule, W, q, F, A, agents, order_seed):
"""Myopic best response of strategic agents, three rounds. agents: dicts with
e, w, sincere, cur, pref(want pass), opts. Mutates F, A, agent['cur']."""
rng = random.Random(order_seed)
for _ in range(3):
order = list(range(len(agents)))
rng.shuffle(order)
changed = False
for k in order:
ag = agents[k]
e, w = ag["e"], ag["w"]
_move(F, A, e, ag["cur"], 0, w)
pick = ag["sincere"]
for o in ag["opts"]:
_move(F, A, e, 0, o, w)
ok = passes(rule, F, A, W, q) == ag["pref"]
_move(F, A, e, o, 0, w)
if ok:
pick = o
break
_move(F, A, e, 0, pick, w)
if pick != ag["cur"]:
changed = True
ag["cur"] = pick
if not changed:
break
def _move(F, A, e, old, new, w):
if old == 1:
F[e] -= w
elif old == -1:
A[e] -= w
if new == 1:
F[e] += w
elif new == -1:
A[e] += w
def run_replicate(cfg, rep, base_seed=20261105):
"""Returns (records, conc). records[k] = (cat_idx, m, d, strat_bits, sincere_bits)."""
pop_rng = random.Random(base_seed * 1_000_003 + rep)
mot_rng = random.Random(base_seed * 7_919 + rep + 17)
estates = [Estate(nm, cfg, pop_rng) for nm in ESTATES]
W = WEIGHTS[cfg.weights]
records, conc = [], []
for _ in range(cfg.motions):
ci = mot_rng.randrange(len(CATS))
q = CATS[ci]
m, d = mot_rng.gauss(0, 1), mot_rng.gauss(0, 1)
zs = [mot_rng.gauss(0, 1) for _ in ESTATES]
sseed = mot_rng.getrandbits(32)
data = [est.motion(ci, m, d, mot_rng, zs[k]) for k, est in enumerate(estates)]
F0, A0 = [0.0] * 4, [0.0] * 4
for e, dt in enumerate(data):
for _, b, w in dt["holders"]:
_move(F0, A0, e, 0, b, w)
conc.append(concentration(data[2]["weights"]) + (data[2]["uncast"], data[2]["delegated"]))
sinc, strat = [], []
for rule in RULES:
sinc.append(passes(rule, F0, A0, W, q))
if cfg.rho <= 0:
strat.append(sinc[-1])
continue
F, A = list(F0), list(A0)
agents = []
for e, (est, dt) in enumerate(zip(estates, data)):
for t, b, w in dt["holders"]:
if est.strategic[t] and not est.faction[t]:
opts = [x for x in (1, -1, 0) if x != b]
agents.append(dict(e=e, w=w, sincere=b, cur=b, pref=dt["u"][t] > 0, opts=[b] + opts))
for i, b in dt["joiners"]:
if est.strategic[i]:
agents.append(dict(e=e, w=1.0, sincere=0, cur=0, pref=dt["u"][i] > 0, opts=[0, b]))
# opts[0] is the sincere option (for joiners: stay out), evaluated first
strategize(rule, W, q, F, A, agents, sseed)
strat.append(passes(rule, F, A, W, q))
records.append((ci, m, d, tuple(strat), tuple(sinc)))
return records, conc
# ---------------------------------------------------------------- metrics
def _mean(xs):
return sum(xs) / len(xs) if xs else float("nan")
def rep_metrics(records, rule_idx, which=3, cats=None):
"""PS, MT, BPP, GBR, pass rate for one replicate and rule. which: 3 strategic, 4 sincere."""
rows = [(m, d, rec[which][rule_idx]) for rec in records
for (ci, m, d) in [rec[:3]] if cats is None or ci in cats]
def diff(sel_pos, sel_neg):
p = [o for (m, d, o) in rows if sel_pos(m, d)]
n = [o for (m, d, o) in rows if sel_neg(m, d)]
return _mean(p) - _mean(n) if p and n else float("nan")
ps = diff(lambda m, d: d > 0, lambda m, d: d < 0)
mt = diff(lambda m, d: m > 0, lambda m, d: m < 0)
bad = [o for (m, d, o) in rows if m < -.5 and d > .5]
good = [o for (m, d, o) in rows if m > .5 and d < -.5]
return dict(PS=ps, MT=mt, BPP=_mean(bad) if bad else float("nan"),
GBR=1 - _mean(good) if good else float("nan"),
PASS=_mean([o for (_, _, o) in rows]) if rows else float("nan"))
def mean_ci(xs):
xs = [x for x in xs if x == x]
if not xs:
return (float("nan"), float("nan"))
mu = _mean(xs)
if len(xs) < 2:
return (mu, float("nan"))
var = sum((x - mu) ** 2 for x in xs) / (len(xs) - 1)
return (mu, 1.96 * math.sqrt(var / len(xs)))
def verdict(dps, dmt):
"""SPEC section 8. dps, dmt: (mean, ci) of PS(UNI)-PS(DIRECT) and MT(UNI)-MT(DIRECT)."""
if dps[0] != dps[0]:
return "n/a"
if (dps[0] - dps[1] > 0) or dmt[0] < -.10:
return "fails"
if dps[0] + dps[1] < 0 and dmt[0] >= -.05:
return "supported"
return "undecided"

View file

@ -0,0 +1,185 @@
"""One-command rerun: python3 -I run.py [--quick] [--out results.md] [--jobs N]
Writes a deterministic markdown results file (no timestamps). See SPEC.md section 7.
"""
import argparse
import os
import sys
from multiprocessing import Pool
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import estatesim as E # noqa: E402
REPS, MOTIONS = 40, 500
QUICK_REPS, QUICK_MOTIONS = 4, 100
METRICS = ("PS", "MT", "BPP", "GBR", "PASS")
PAIRS = (("UNI", "DIRECT"), ("WTA", "DIRECT"), ("UNI", "WTA"))
def _job(args):
cfg, rep = args
return E.run_replicate(cfg, rep)
def evaluate(cfg_list, reps, jobs):
"""Run every (cfg, rep); returns list of per-config result dicts."""
tasks = [(cfg, r) for cfg in cfg_list for r in range(reps)]
if jobs > 1:
with Pool(jobs) as p:
out = p.map(_job, tasks, chunksize=2)
else:
out = [_job(t) for t in tasks]
res = []
for k, cfg in enumerate(cfg_list):
chunk = out[k * reps:(k + 1) * reps]
res.append(summarize(cfg, chunk))
return res
def summarize(cfg, chunk):
recs = [c[0] for c in chunk]
per = {} # rule -> metric -> list over reps (strategic) ; plus sincere PS
for ri, rule in enumerate(E.RULES):
per[rule] = {m: [] for m in METRICS}
per[rule]["PS_sincere"] = []
per[rule]["FLIP"] = []
for rec in recs:
mm = E.rep_metrics(rec, ri)
for m in METRICS:
per[rule][m].append(mm[m])
per[rule]["PS_sincere"].append(E.rep_metrics(rec, ri, which=4)["PS"])
per[rule]["FLIP"].append(sum(1 for r in rec if r[3][ri] != r[4][ri]) / len(rec))
percat = {}
for ci in range(len(E.CATS)):
for ri, rule in enumerate(E.RULES):
percat[(ci, rule)] = [E.rep_metrics(rec, ri, cats={ci})["PS"] for rec in recs]
paired = {}
for a, b in PAIRS:
for m in ("PS", "MT"):
paired[(a, b, m)] = E.mean_ci([x - y for x, y in zip(per[a][m], per[b][m])])
conc = {}
allc = [c for ch in chunk for c in ch[1]]
for k, nm in enumerate(("top_share", "gini", "n_eff", "weight")):
conc[nm] = sum(c[k] for c in allc) / len(allc)
dl = sum(c[5] for c in allc)
conc["uncast"] = sum(c[4] for c in allc) / dl if dl else 0.0 # share of delegated votes left uncast
return dict(cfg=cfg, per=per, percat=percat, paired=paired, conc=conc,
verdict=E.verdict(paired[("UNI", "DIRECT", "PS")], paired[("UNI", "DIRECT", "MT")]))
def f(x, ci=None, d=3):
if x != x:
return "n/a"
s = f"{x:.{d}f}"
return s + (f" ±{ci:.{d}f}" if ci is not None and ci == ci else "")
def rule_table(r):
lines = ["| rule | PS | MT | PI=PS/MT | BPP | GBR | pass | PS sincere | outcome flipped by strategy |",
"|---|---|---|---|---|---|---|---|---|"]
for rule in E.RULES:
p = r["per"][rule]
ps, mt = E.mean_ci(p["PS"]), E.mean_ci(p["MT"])
pi = ps[0] / mt[0] if mt[0] >= .05 else float("nan")
lines.append(f"| {rule} | {f(*ps)} | {f(*mt)} | {f(pi)} | {f(*E.mean_ci(p['BPP']))} | "
f"{f(*E.mean_ci(p['GBR']))} | {f(*E.mean_ci(p['PASS']))} | "
f"{f(*E.mean_ci(p['PS_sincere']))} | {f(*E.mean_ci(p['FLIP']))} |")
return lines
def pair_line(r):
pr = r["paired"]
parts = []
for a, b in PAIRS:
parts.append(f"{a}-{b}: dPS {f(*pr[(a, b, 'PS')])}, dMT {f(*pr[(a, b, 'MT')])}")
return "; ".join(parts) + f"; claim (UNI vs DIRECT): **{r['verdict']}**"
def percat_table(r):
lines = ["| q | " + " | ".join(E.RULES) + " |", "|---|" + "---|" * len(E.RULES)]
for ci, q in enumerate(E.CATS):
lines.append(f"| {q} | " + " | ".join(f(*E.mean_ci(r["percat"][(ci, ru)]), d=2) for ru in E.RULES) + " |")
return lines
def experiments(quick):
X = []
if quick:
alphas, caps, facs, betas = (0.0, 1.5), (.2, None), (0, 20), (0.0, 3.0)
else:
alphas, caps, facs, betas = (0.0, .5, 1.0, 1.5, 2.5, 4.0), (.10, .15, .20, .25, .30, None), (0, 10, 20, 40), (0.0, 1.0, 3.0)
X.append(("E1 baseline (sincere, no delegation, no faction)",
[(w, E.Config(weights=w)) for w in ("equal", "minds-heavy")], True))
X.append(("E2 strategic agents (rho=.25) and Users faction size",
[(f"faction={k}", E.Config(rho=.25, faction=k)) for k in facs], True))
X.append(("E3 delegation: concentration alpha x per-delegate cap (depth 3, strategic, faction 20)",
[(f"alpha={a} cap={c}", E.Config(delegation=True, alpha=a, cap=c, rho=.25, faction=20))
for a in alphas for c in caps], False))
X.append(("E3b same grid, no faction, sincere (isolates delegation)",
[(f"alpha={a} cap={c}", E.Config(delegation=True, alpha=a, cap=c))
for a in alphas for c in caps], False))
depths = (2, None) if quick else (2, 3, 4, 6, None)
X.append(("E3c transitive depth x delegate turnout x onward delegation (alpha=1.5, cap=.2, strategic, faction 20)",
[(f"depth={d} dturn={t} onward={o}", E.Config(delegation=True, alpha=1.5, depth=d, dturn=t, onward=o, cap=.2, rho=.25, faction=20))
for d in depths for t in (.95, .6, .3) for o in (.3, .6)], False))
X.append(("E3d cap basis: share of cast weight vs share of roll (alpha=1.5, depth=3, strategic, faction 20)",
[(f"basis={bs} cap={c}", E.Config(delegation=True, alpha=1.5, cap=c, cap_basis=bs, rho=.25, faction=20))
for bs in ("cast", "roll") for c in ((.1, .2) if quick else (.05, .1, .2, .3))], False))
X.append(("E4 appeal attraction beta (alpha=1.5, cap=.2, strategic, faction 20)",
[(f"beta={b}", E.Config(delegation=True, alpha=1.5, cap=.2, beta=b, rho=.25, faction=20))
for b in betas], True))
rob = [("users appeal b=.4", (("Users", "b", .4),)), ("users appeal b=.9 (default)", ()),
("users appeal b=1.4", (("Users", "b", 1.4),)),
("users turnout .05", (("Users", "turnout", .05),)), ("users turnout .30", (("Users", "turnout", .30),)),
("minds shared s=.3", (("Minds", "s", .3),)), ("minds shared s=1.5", (("Minds", "s", 1.5),))]
X.append(("E5 robustness to assumed parameters (no delegation, strategic, faction 20)",
[(nm, E.Config(rho=.25, faction=20, overrides=ov)) for nm, ov in rob], True))
X.append(("E5b same, with delegation (alpha=1.5, cap=.2)",
[(nm, E.Config(delegation=True, alpha=1.5, cap=.2, rho=.25, faction=20, overrides=ov))
for nm, ov in rob], False))
return X
def main(argv=None):
ap = argparse.ArgumentParser()
ap.add_argument("--quick", action="store_true")
ap.add_argument("--out", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "results.md"))
ap.add_argument("--jobs", type=int, default=min(os.cpu_count() or 1, 16))
a = ap.parse_args(argv)
reps, motions = (QUICK_REPS, QUICK_MOTIONS) if a.quick else (REPS, MOTIONS)
L = [f"# Estate-rule sim results ({'quick' if a.quick else 'full'}: {reps} replicates x {motions} motions per cell)",
"", "Generated by `run.py`; spec in SPEC.md. +/- is a 95% interval over replicates. PS populism sensitivity, "
"MT merit tracking, BPP bad-popular pass rate, GBR good-unpopular block rate (SPEC section 3).", ""]
tally = {"supported": 0, "fails": 0, "undecided": 0, "n/a": 0}
failing = []
for title, cells, detail in experiments(a.quick):
cfgs = [E.Config(**{**c.__dict__, "motions": motions}) for _, c in cells]
res = evaluate(cfgs, reps, a.jobs)
L += [f"## {title}", ""]
summ = ["| cell | dPS UNI-DIRECT | dMT UNI-DIRECT | dPS WTA-DIRECT | PS DIRECT | PS WTA | PS UNI | Users top share | Users n_eff | uncast | claim |",
"|---|---|---|---|---|---|---|---|---|---|---|"]
for (name, _), r in zip(cells, res):
tally[r["verdict"]] += 1
if r["verdict"] == "fails":
failing.append(f"{title.split(' ')[0]} {name}")
pr, pe = r["paired"], r["per"]
summ.append(f"| {name} | {f(*pr[('UNI','DIRECT','PS')])} | {f(*pr[('UNI','DIRECT','MT')])} | "
f"{f(*pr[('WTA','DIRECT','PS')])} | {f(*E.mean_ci(pe['DIRECT']['PS']))} | "
f"{f(*E.mean_ci(pe['WTA']['PS']))} | {f(*E.mean_ci(pe['UNI']['PS']))} | "
f"{r['conc']['top_share']:.3f} | {r['conc']['n_eff']:.1f} | {r['conc']['uncast']:.2f} | {r['verdict']} |")
L += summ + [""]
if detail:
for (name, _), r in zip(cells, res):
L += [f"### {title.split(' ')[0]} / {name}", ""] + rule_table(r) + [""]
L += [pair_line(r), "", "PS by category threshold:", ""] + percat_table(r) + [""]
L += ["## Claim tally over all cells (SPEC section 8)", "",
f"supported {tally['supported']}, fails {tally['fails']}, undecided {tally['undecided']}, n/a {tally['n/a']}", ""]
if failing:
L += ["Failing cells:", ""] + [f"- {x}" for x in failing] + [""]
with open(a.out, "w") as fh:
fh.write("\n".join(L) + "\n")
print(f"wrote {a.out}: {tally}")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,288 @@
"""python3 -I test_estatesim.py"""
import os
import random
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import estatesim as E # noqa: E402
W = (25, 25, 25, 25)
def est(cap=None, name="Users"):
cfg = E.Config(delegation=True, cap=cap)
e = E.Estate(name, cfg, random.Random(1))
e.target = [None] * e.n
e.scope = [None] * e.n
return e
def edge(e, i, t, scope=(0, 1, 2, 3, 4)):
e.target[i], e.scope[i] = t, frozenset(scope)
class RuleTests(unittest.TestCase):
def test_threshold_tie_passes(self):
# one live estate, exactly q For
self.assertTrue(E.passes("DIRECT", [.5, 0, 0, 0], [.5, 0, 0, 0], W, .5))
self.assertFalse(E.passes("DIRECT", [.4, 0, 0, 0], [.6, 0, 0, 0], W, .5))
def test_wta_tie_drops_estate_out(self):
# Minds tie drops out; Members 3-1 For decides alone
self.assertTrue(E.passes("WTA", [4, 3, 0, 0], [4, 1, 0, 0], W, .66))
# tied estate alone casts nothing: nothing passes
self.assertFalse(E.passes("WTA", [4, 0, 0, 0], [4, 0, 0, 0], W, .15))
def test_uni_uses_category_threshold_inside_estate(self):
# 40% For inside each estate: UNI at q=.3 casts all For, WTA casts all Against
F, A = [4, 4, 4, 4], [6, 6, 6, 6]
self.assertTrue(E.passes("UNI", F, A, W, .3))
self.assertFalse(E.passes("WTA", F, A, W, .3))
self.assertFalse(E.passes("UNI", F, A, W, .5))
def test_uni_two_of_four_blocs_at_half(self):
self.assertTrue(E.passes("UNI", [9, 9, 1, 1], [1, 1, 9, 9], W, .5)) # 50% of weight
self.assertFalse(E.passes("UNI", [9, 1, 1, 1], [1, 9, 9, 9], W, .5))
def test_renormalizes_over_live_estates(self):
self.assertTrue(E.passes("UNI", [5, 0, 0, 0], [1, 0, 0, 0], W, .66)) # empty estates not counted
self.assertFalse(E.passes("DIRECT", [0] * 4, [0] * 4, W, .15))
def test_head_ignores_estate_weights(self):
self.assertTrue(E.passes("HEAD", [1, 0, 9, 0], [9, 0, 0, 0], W, .5))
self.assertFalse(E.passes("DIRECT", [1, 0, 9, 0], [9, 0, 0, 0], W, .6))
class DelegationTests(unittest.TestCase):
def setUp(self):
self.e = est()
self.n = self.e.n
def voting(self, ids):
v = [False] * self.n
for i in ids:
v[i] = True
return v
def test_delegated_weight_goes_to_delegate(self):
edge(self.e, 1, 0)
w, dest = self.e.resolve(0, self.voting([0]))
self.assertEqual(w, {0: 2.0})
self.assertEqual(dest, {1: 0})
def test_direct_vote_overrides_delegation(self):
edge(self.e, 1, 0)
w, dest = self.e.resolve(0, self.voting([0, 1]))
self.assertEqual(w, {0: 1.0, 1: 1.0})
self.assertEqual(dest, {})
def test_scope_out_of_category_is_ignored(self):
edge(self.e, 1, 0, scope=(2,))
w, _ = self.e.resolve(0, self.voting([0]))
self.assertEqual(w, {0: 1.0})
w, _ = self.e.resolve(2, self.voting([0]))
self.assertEqual(w, {0: 2.0})
def chain_estate(self, depth, n_links):
e = est()
e.cfg = E.Config(delegation=True, depth=depth)
for k in range(1, n_links + 1): # n_links -> ... -> 1 -> 0
edge(e, k, k - 1)
return e
def test_chain_within_depth_reaches_holder(self):
e = self.chain_estate(2, 2) # 2 -> 1 -> 0, two edges
w, dest = e.resolve(0, self.voting([0]))
self.assertEqual(w, {0: 3.0})
self.assertEqual(dest, {1: 0, 2: 0})
def test_over_depth_is_uncast_not_redirected(self):
e = self.chain_estate(2, 3) # 3 needs three edges
w, dest = e.resolve(0, self.voting([0]))
self.assertNotIn(3, dest)
self.assertEqual(w, {0: 3.0})
self.assertEqual(e.last_uncast, 1)
def test_unlimited_depth_follows_long_chain(self):
e = self.chain_estate(None, 9)
w, _ = e.resolve(0, self.voting([0]))
self.assertEqual(w, {0: 10.0})
def test_intermediate_direct_vote_terminates_chain(self):
e = self.chain_estate(None, 3) # 3 -> 2 -> 1 -> 0 ; 1 votes itself
w, dest = e.resolve(0, self.voting([0, 1]))
self.assertEqual(w, {0: 1.0, 1: 3.0})
self.assertEqual(dest, {2: 1, 3: 1})
def test_loop_is_uncast(self):
e = est()
e.cfg = E.Config(delegation=True, depth=None)
edge(e, 1, 2)
edge(e, 2, 1)
edge(e, 3, 1) # feeds the loop: also void
w, dest = e.resolve(0, self.voting([0]))
self.assertEqual((w, dest), ({0: 1.0}, {}))
self.assertEqual(e.last_uncast, 3)
def test_scope_is_overlap_of_all_links(self):
e = est()
e.cfg = E.Config(delegation=True, depth=None)
edge(e, 2, 1, scope=(0, 1)) # 2 -> 1 -> 0
edge(e, 1, 0, scope=(1, 2))
for cat, expect in ((0, {}), (1, {2: 0, 1: 0}), (2, {1: 0})):
_, dest = e.resolve(cat, self.voting([0]))
self.assertEqual(dest, expect, cat)
def test_dead_end_is_uncast(self):
e = est()
edge(e, 2, 1) # 1 does not vote and delegates nowhere
w, dest = e.resolve(0, self.voting([0]))
self.assertEqual((w, dest), ({0: 1.0}, {}))
self.assertEqual(e.last_uncast, 1)
def test_cap_voids_excess_and_floors_at_one(self):
e = est(cap=.5) # cast weight 12, cap 6
for i in range(1, 11):
edge(e, i, 0)
voting = [i in (0, 20) for i in range(e.n)]
w, _ = e.resolve(0, voting)
self.assertEqual(w, {0: 6.0, 20: 1.0}) # 11 delegated+own capped to 6, excess void
e = est(cap=.001, name="Patrons") # below one voter: floor is 1
w, _ = e.resolve(0, [i == 0 for i in range(e.n)])
self.assertEqual(w, {0: 1.0})
def test_cap_applies_to_final_holder_not_each_link(self):
e = est(cap=.4)
e.cfg = E.Config(delegation=True, depth=None, cap=.4)
for i in range(1, 6):
edge(e, i, 6) # five delegators to 6 ...
edge(e, 6, 0) # ... who delegates to 0
w, _ = e.resolve(0, self.voting([0, 20])) # 0 holds 1+6, cast 8 -> cap 3.2
self.assertEqual(w[0], 3.2)
def test_cap_basis_roll(self):
e = est()
e.cfg = E.Config(delegation=True, cap=.1, cap_basis="roll") # 300 * .1 = 30
for i in range(1, 41):
edge(e, i, 0)
w, _ = e.resolve(0, self.voting([0]))
self.assertEqual(w, {0: 30.0})
def contradicting_estate(self):
e = est()
edge(e, 1, 0)
e.turn[0], e.kap[0], e.a[0], e.b[0], e.sig[0] = 1.0, 0.0, 1.0, 0.0, .01
e.turn[1], e.kap[1], e.a[1], e.b[1], e.sig[1] = 0.0, 0.0, -3.0, 0.0, .01
return e
def test_contradiction_forces_direct_vote_then_revokes_at_second_strike(self):
e = self.contradicting_estate()
dt = e.motion(0, 1.0, 0.0, random.Random(1), 0.0)
self.assertIn(1, [t for t, _, _ in dt["holders"]]) # overrides this motion
self.assertEqual(e.target[1], 0) # one strike: still delegated
self.assertEqual(e.strikes[1], 1)
e.motion(0, 1.0, 0.0, random.Random(2), 0.0)
self.assertIsNone(e.target[1]) # second strike revokes
e.motion(0, 1.0, 0.0, random.Random(3), 0.0)
self.assertEqual(e.strikes[1], 2) # revoked: no further strikes, never returns
self.assertIsNone(e.target[1])
def test_direct_voters_never_forwarded(self):
edge(self.e, 1, 0)
w, _ = self.e.resolve(0, self.voting([0, 1]))
self.assertNotIn(2, w)
self.assertEqual(sum(w.values()), 2.0)
class StrategyTests(unittest.TestCase):
def test_deviation_never_makes_agent_worse_off(self):
rng = random.Random(3)
for _ in range(300):
q = rng.choice(E.CATS)
F = [float(rng.randrange(0, 6)) for _ in range(4)]
A = [float(rng.randrange(0, 6)) for _ in range(4)]
rule = rng.choice(E.RULES)
e = rng.randrange(4)
for b in (1, -1):
F0, A0 = list(F), list(A)
E._move(F0, A0, e, 0, b, 1.0)
pref = b == 1
ag = dict(e=e, w=1.0, sincere=b, cur=b, pref=pref, opts=[b] + [x for x in (1, -1, 0) if x != b])
sincere_ok = E.passes(rule, F0, A0, W, q) == pref
E.strategize(rule, W, q, F0, A0, [ag], 7)
if sincere_ok:
self.assertTrue(E.passes(rule, F0, A0, W, q) == pref)
def test_strategic_joiner_joins_only_when_pivotal(self):
F, A = [3.0, 0, 0, 0], [3.0, 0, 0, 0] # tie, q=.5 passes under UNI already
ag = dict(e=0, w=1.0, sincere=0, cur=0, pref=True, opts=[0, 1])
E.strategize("UNI", W, .5, F, A, [ag], 1)
self.assertEqual(ag["cur"], 0) # already passes: stays home
F, A = [2.0, 0, 0, 0], [3.0, 0, 0, 0] # 40% For: fails at .5, joining makes 3/6
ag = dict(e=0, w=1.0, sincere=0, cur=0, pref=True, opts=[0, 1])
E.strategize("UNI", W, .5, F, A, [ag], 1)
self.assertEqual(ag["cur"], 1)
def test_faction_votes_agenda_regardless_of_preference(self):
cfg = E.Config(faction=20)
e = E.Estate("Users", cfg, random.Random(2))
self.assertEqual(sum(e.faction), 20)
dt = e.motion(0, -2.0, 2.0, random.Random(9), 0.0) # d - m > 0: faction votes For
ballots = {t: b for t, b, _ in dt["holders"]}
for i in range(e.n):
if e.faction[i]:
self.assertEqual(ballots[i], 1)
class HarnessTests(unittest.TestCase):
CFG = E.Config(delegation=True, alpha=1.5, cap=.2, rho=.25, faction=10, motions=30)
def test_deterministic(self):
self.assertEqual(E.run_replicate(self.CFG, 3), E.run_replicate(self.CFG, 3))
def test_draws_paired_across_rules_and_configs(self):
a, _ = E.run_replicate(E.Config(motions=30), 1)
b, _ = E.run_replicate(self.CFG, 1)
self.assertEqual([r[:3] for r in a], [r[:3] for r in b]) # same (category, merit, appeal)
def test_sincere_run_has_no_strategy_flips(self):
r, _ = E.run_replicate(E.Config(motions=60), 0)
self.assertTrue(all(x[3] == x[4] for x in r))
def test_metrics_on_known_outcomes(self):
# outcome == (d > 0): PS = 1, MT ~ 0 (independent), BPP = 1, GBR = 1 - 0 = 1
rng = random.Random(1)
recs = []
for _ in range(4000):
m, d = rng.gauss(0, 1), rng.gauss(0, 1)
recs.append((0, m, d, (d > 0,) * 4, (d > 0,) * 4))
mm = E.rep_metrics(recs, 0)
self.assertEqual(mm["PS"], 1.0)
self.assertLess(abs(mm["MT"]), .1)
self.assertEqual(mm["BPP"], 1.0)
self.assertEqual(mm["GBR"], 1.0)
# outcome == (m > 0): PS ~ 0, MT = 1, BPP = 0, GBR = 0
recs = [(0, m, d, (m > 0,) * 4, (m > 0,) * 4) for (_, m, d, _, _) in recs]
mm = E.rep_metrics(recs, 0)
self.assertEqual(mm["MT"], 1.0)
self.assertEqual(mm["BPP"], 0.0)
self.assertEqual(mm["GBR"], 0.0)
self.assertLess(abs(mm["PS"]), .1)
def test_concentration(self):
top, gini, neff, tot = E.concentration([1, 1, 1, 1])
self.assertEqual((top, round(gini, 9), neff, tot), (.25, 0.0, 4.0, 4))
top, gini, neff, _ = E.concentration([1, 1, 1, 97])
self.assertGreater(gini, .7)
self.assertLess(neff, 1.1)
def test_verdict_rules(self):
self.assertEqual(E.verdict((-.1, .02), (0.0, .02)), "supported")
self.assertEqual(E.verdict((.1, .02), (0.0, .02)), "fails")
self.assertEqual(E.verdict((-.1, .02), (-.12, .02)), "fails") # dampened by not working
self.assertEqual(E.verdict((-.01, .05), (0.0, .02)), "undecided")
if __name__ == "__main__":
unittest.main()