diff --git a/docs/research/2026-11-05-estate-rule-sim/estatesim.py b/docs/research/2026-11-05-estate-rule-sim/estatesim.py new file mode 100644 index 0000000..19e55ac --- /dev/null +++ b/docs/research/2026-11-05-estate-rule-sim/estatesim.py @@ -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" diff --git a/docs/research/2026-11-05-estate-rule-sim/run.py b/docs/research/2026-11-05-estate-rule-sim/run.py new file mode 100644 index 0000000..3ff5c34 --- /dev/null +++ b/docs/research/2026-11-05-estate-rule-sim/run.py @@ -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() diff --git a/docs/research/2026-11-05-estate-rule-sim/test_estatesim.py b/docs/research/2026-11-05-estate-rule-sim/test_estatesim.py new file mode 100644 index 0000000..2437b00 --- /dev/null +++ b/docs/research/2026-11-05-estate-rule-sim/test_estatesim.py @@ -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()