#!/usr/bin/env python3 """Igneum sustained-mining finality: vote-weight simulation. Rule under test (from the design doc, finality section): Each miner key's vote weight is the sum over the trailing WINDOW days of its COUNTED blue blocks. A day's counted blocks are counted[d] = min(actual[d], GROWTH * counted[d-1] + FLOOR) so a key's counted output can at most double day over day (GROWTH = 2), plus a small FLOOR so a key with no history can start at all. A checkpoint locks when the signing keys hold two thirds of TOTAL weight. Model assumptions (all of them, stated once here and repeated in results.md): * Time step is one day. The network produces 86,400 blocks a day (1 block/s). Each key's blocks that day are Poisson with mean 86,400 x its hashrate share. Difficulty is assumed to retarget perfectly, so total blocks stay at 86,400 whatever the total hashrate does. * Every block a key produces is a blue block (no DAG, no red blocks, no latency). * Block rewards are unweighted: a key's reward share is its block share. * Honest keys sign every checkpoint they are sampled for. An attacker's keys sign only their own checkpoints. "Can lock alone" means that group's weight is at least two thirds of total weight. * The alternative "active total" (scenario E) counts a key as active if it was sampled into at least one checkpoint committee in the last 24 hours. We approximate the number of times a key is sampled in a day as Poisson with mean CHECKPOINTS_PER_DAY x COMMITTEE x (its weight share). * No VRF sampling noise on the lock itself, no network latency, no reorgs. Standard library plus numpy. Deterministic for a given --seed. Usage: python3 finality_sim.py # all scenarios, all floors, markdown to stdout python3 finality_sim.py --scenarios B,C --floors 1,100 python3 finality_sim.py --seed 11 --growth 2 --window 30 """ import argparse import sys import numpy as np BLOCKS_PER_DAY = 86_400 CHECKPOINTS_PER_DAY = 2_880 # one checkpoint every 30 s TWO_THIRDS = 2.0 / 3.0 WARM_DAYS = 60 # honest-only days before any event in B to F PARETO_SHAPE = 1.0 # shape 1 gives top pool ~17%, top 10 ~50%, bottom 500 keys ~8% N_HONEST = 1_000 class Params: default_presence = 0 # set from --presence in main() def __init__(self, floor, growth=2.0, window=30, committee=100, presence=None): self.floor = float(floor) self.growth = float(growth) self.window = int(window) self.committee = int(committee) # presence gate (proposed addition, off by default): a key's daily cap stays at FLOOR # until it has produced at least one counted block on PRESENCE of the WINDOW days. self.presence = int(Params.default_presence if presence is None else presence) class Network: """Keys, their hashrate, online flag, group label, and the counted-block window.""" def __init__(self, params, rng): self.p = params self.rng = rng self.hash = np.zeros(0) self.online = np.zeros(0, dtype=bool) self.group = np.zeros(0, dtype=np.int64) self.window = np.zeros((self.p.window, 0)) self.prev = np.zeros(0) self.last_actual = np.zeros(0) self.ptr = 0 self.day = 0 def add_keys(self, hashrates, group): hashrates = np.asarray(hashrates, dtype=float) n = hashrates.size self.hash = np.concatenate([self.hash, hashrates]) self.online = np.concatenate([self.online, np.ones(n, dtype=bool)]) self.group = np.concatenate([self.group, np.full(n, group, dtype=np.int64)]) self.window = np.concatenate([self.window, np.zeros((self.p.window, n))], axis=1) self.prev = np.concatenate([self.prev, np.zeros(n)]) self.last_actual = np.concatenate([self.last_actual, np.zeros(n)]) def set_hash(self, mask, value): self.hash[mask] = value def step(self): h = np.where(self.online, self.hash, 0.0) share = h / h.sum() actual = self.rng.poisson(BLOCKS_PER_DAY * share).astype(float) cap = np.floor(self.p.growth * self.prev) + self.p.floor if self.p.presence > 0: present_days = (self.window > 0).sum(axis=0) cap = np.where(present_days >= self.p.presence, cap, self.p.floor) counted = np.minimum(actual, cap) self.window[self.ptr] = counted self.ptr = (self.ptr + 1) % self.p.window self.prev = counted self.last_actual = actual self.day += 1 def run(self, days): for _ in range(days): self.step() def weight(self): return self.window.sum(axis=0) def group_mask(self, g): return self.group == g def group_weight_share(self, g, total_mask=None): w = self.weight() if total_mask is None: total = w.sum() else: total = w[total_mask].sum() if total <= 0: return 0.0 return w[self.group_mask(g)].sum() / total def group_block_share(self, g): a = self.last_actual return a[self.group_mask(g)].sum() / max(a.sum(), 1.0) def active_mask(self): """Keys that signed at least one checkpoint in the last 24 h (approximation, see header).""" w = self.weight() total = w.sum() if total <= 0: return np.zeros_like(self.online) lam = CHECKPOINTS_PER_DAY * self.p.committee * (w / total) sampled = self.rng.poisson(lam) >= 1 return self.online & sampled & (w > 0) def pareto_hashrates(rng, n, total=1.0): h = rng.pareto(PARETO_SHAPE, n) + 1.0 return h * (total / h.sum()) def gini(x): x = np.sort(np.asarray(x, dtype=float)) n = x.size if n == 0 or x.sum() == 0: return 0.0 cum = np.cumsum(x) return (n + 1 - 2.0 * cum.sum() / cum[-1]) / n def fmt_pct(x): return "%.1f%%" % (100.0 * x) def fmt_day(d): return "never" if d is None else str(d) def md_table(headers, rows): out = ["| " + " | ".join(headers) + " |", "|" + "---|" * len(headers)] for r in rows: out.append("| " + " | ".join(str(c) for c in r) + " |") return "\n".join(out) def first_day(series, pred): """series: list of (t, value). Return first t where pred(value), else None.""" for t, v in series: if pred(v): return t return None # ---------------------------------------------------------------- scenarios def warm_network(params, rng, days=WARM_DAYS): net = Network(params, rng) net.add_keys(pareto_hashrates(rng, N_HONEST), group=0) net.run(days) return net def scenario_a(floors, seed, growth, window, committee): """Steady state: weight vs hashrate after 60 honest days.""" rows = [] ramp_rows = [] for f in floors: rng = np.random.default_rng(seed) p = Params(f, growth, window, committee) net = Network(p, rng) net.add_keys(pareto_hashrates(rng, N_HONEST), group=0) full = BLOCKS_PER_DAY * window reached = None for d in range(1, WARM_DAYS + 1): net.step() if reached is None and net.weight().sum() >= 0.99 * full: reached = d w = net.weight() ws = w / w.sum() hs = net.hash / net.hash.sum() order = np.argsort(-hs) corr = np.corrcoef(hs, ws)[0, 1] rel = ws / hs under = int((rel < 0.9).sum()) rows.append([ int(f), fmt_pct(w.sum() / full), "%.5f" % corr, "%.3f / %.3f" % (gini(hs), gini(ws)), fmt_pct(hs[order[0]]) + " / " + fmt_pct(ws[order[0]]), fmt_pct(hs[order[:10]].sum()) + " / " + fmt_pct(ws[order[:10]].sum()), fmt_pct(hs[order[500:]].sum()) + " / " + fmt_pct(ws[order[500:]].sum()), "%.3f / %.3f" % (rel.min(), rel.max()), under, ]) ramp_rows.append([int(f), fmt_day(reached)]) headers = ["floor f", "total weight / full", "corr(hash, weight)", "Gini hash / weight", "top-1 hash / weight", "top-10 hash / weight", "bottom-500 hash / weight", "min / max weight:hash ratio", "keys under 0.9x"] out = ["### A. Steady state, 1,000 honest keys, day 60", "", md_table(headers, rows), "", "Day the network's total weight first reached 99% of the full window (30 x 86,400 = 2,592,000):", "", md_table(["floor f", "day"], ramp_rows)] return "\n".join(out) def run_attack(params, rng, attacker_mult, n_keys, trickle_days, report_days): """Honest warm-up, optional trickle, then burst. Returns per-day series for the attacker group (1).""" net = Network(params, rng) net.add_keys(pareto_hashrates(rng, N_HONEST), group=0) burst_blocks = BLOCKS_PER_DAY * attacker_mult / (1.0 + attacker_mult) trickle_info = None if trickle_days > 0: net.run(WARM_DAYS - trickle_days) per_key = min(params.floor, burst_blocks / n_keys) target = per_key * n_keys # attacker blocks per day during trickle a_hash = target / (BLOCKS_PER_DAY - target) # in units of honest hashrate (= 1.0) net.add_keys(np.full(n_keys, a_hash / n_keys), group=1) net.run(trickle_days) trickle_info = dict(per_key=per_key, total=target, share=target / BLOCKS_PER_DAY, weight_share_at_burst=net.group_weight_share(1)) else: net.run(WARM_DAYS) net.add_keys(np.zeros(n_keys), group=1) # burst net.set_hash(net.group_mask(1), attacker_mult / n_keys) series = [] block_share_day1 = None for t in range(1, report_days + 1): net.step() if t == 1: block_share_day1 = net.group_block_share(1) series.append((t, net.group_weight_share(1))) return series, block_share_day1, trickle_info def scenario_b(floors, seed, growth, window, committee, mults=(1.5, 3.0), report_days=46): out = [] pick = [1, 2, 3, 5, 7, 10, 14, 20, 25, 30, 35, 40, 46] for mult in mults: share = mult / (1.0 + mult) label = "attacker hashrate = %.1fx honest (%s of network)" % (mult, fmt_pct(share)) if mult != 1.5: label += ", supplementary run, not in the brief" results = {} for f in floors: rng = np.random.default_rng(seed) p = Params(f, growth, window, committee) results[f] = run_attack(p, rng, mult, 1, 0, report_days) rows = [] for t in pick: rows.append(["+%d" % t] + [fmt_pct(dict(results[f][0])[t]) for f in floors]) headers = ["day after burst"] + ["f=%d" % f for f in floors] cross = [ ["crosses 1/3 (honest keys can no longer lock)"] + [fmt_day(first_day(results[f][0], lambda v: v > 1.0 - TWO_THIRDS)) for f in floors], ["crosses 50%"] + [fmt_day(first_day(results[f][0], lambda v: v > 0.5)) for f in floors], ["crosses 2/3"] + [fmt_day(first_day(results[f][0], lambda v: v >= TWO_THIRDS)) for f in floors], ["max share in %d days" % report_days] + [fmt_pct(max(v for _, v in results[f][0])) for f in floors], ["block-reward share, burst day 1"] + [fmt_pct(results[f][1]) for f in floors], ] out.append("### B. Rental burst at day 60, one key, %s" % label) out.append("") out.append("Attacker weight share of total:") out.append("") out.append(md_table(headers, rows)) out.append("") out.append(md_table(["event"] + ["f=%d" % f for f in floors], cross)) out.append("") return "\n".join(out) def scenario_c(floors, seed, growth, window, committee, ks=(100, 1_000, 10_000), mults=(1.5, 3.0), report_days=46): out = [] for mult in mults: share = mult / (1.0 + mult) label = "attacker hashrate = %.1fx honest (%s of network)" % (mult, fmt_pct(share)) if mult != 1.5: label += ", supplementary run" # baseline B for comparison base = {} for f in floors: rng = np.random.default_rng(seed) base[f] = run_attack(Params(f, growth, window, committee), rng, mult, 1, 0, report_days)[0] rows_13, rows_50, rows_23, rows_cost, rows_d1 = [], [], [], [], [] one_third = 1.0 - TWO_THIRDS def add_rows(name, series_by_f): rows_13.append([name] + [fmt_day(first_day(series_by_f[f], lambda v: v > one_third)) for f in floors]) rows_50.append([name] + [fmt_day(first_day(series_by_f[f], lambda v: v > 0.5)) for f in floors]) rows_23.append([name] + [fmt_day(first_day(series_by_f[f], lambda v: v >= TWO_THIRDS)) for f in floors]) rows_d1.append([name] + [fmt_pct(dict(series_by_f[f])[1]) for f in floors]) add_rows("B: 1 key, no trickle", base) # C0: K fresh keys appear on the burst day with no trickle (cheapest split, costs only key creation) for k in list(ks) + [50_000]: res = {} for f in floors: rng = np.random.default_rng(seed) res[f] = run_attack(Params(f, growth, window, committee), rng, mult, k, 0, report_days)[0] add_rows("C0: K=%d, no trickle" % k, res) # C: K keys trickle at the floor for 30 days, then flood (the brief's scenario) for k in ks: res = {} for f in floors: rng = np.random.default_rng(seed) res[f] = run_attack(Params(f, growth, window, committee), rng, mult, k, 30, report_days) add_rows("C: K=%d, 30-day trickle" % k, {f: res[f][0] for f in floors}) rows_cost.append(["K=%d" % k] + [ "%.4g blk/key/day, %s of network, %s weight at burst" % ( res[f][2]["per_key"], fmt_pct(res[f][2]["share"]), fmt_pct(res[f][2]["weight_share_at_burst"])) for f in floors]) headers = ["run"] + ["f=%d" % f for f in floors] out.append("### C. Key splitting, %s" % label) out.append("") out.append("Attacker weight share one day after the burst:") out.append("") out.append(md_table(headers, rows_d1)) out.append("") out.append("Days after the burst until the attacker crosses 1/3 of total weight (honest keys can no longer lock):") out.append("") out.append(md_table(headers, rows_13)) out.append("") out.append("Days after the burst until the attacker crosses 50% of total weight:") out.append("") out.append(md_table(headers, rows_50)) out.append("") out.append("Days after the burst until the attacker reaches 2/3 of total weight:") out.append("") out.append(md_table(headers, rows_23)) out.append("") out.append("Trickle phase (days 30 to 60): per-key rate, attacker share of network blocks, " "attacker weight share on the burst day:") out.append("") out.append(md_table(["run"] + ["f=%d" % f for f in floors], rows_cost)) out.append("") return "\n".join(out) def scenario_d(floors, seed, growth, window, committee, report_days=46): pick = [1, 3, 5, 10, 15, 20, 25, 30, 35, 46] results = {} for f in floors: rng = np.random.default_rng(seed) p = Params(f, growth, window, committee) net = warm_network(p, rng) net.add_keys(pareto_hashrates(rng, N_HONEST, total=1.0), group=2) # new honest cohort, same total hashrate series_new, series_old = [], [] for t in range(1, report_days + 1): net.step() series_new.append((t, net.group_weight_share(2))) series_old.append((t, net.group_weight_share(0))) results[f] = (series_new, series_old) rows = [] for t in pick: rows.append(["+%d" % t] + [fmt_pct(dict(results[f][0])[t]) for f in floors]) headers = ["day after doubling"] + ["f=%d" % f for f in floors] ev = [ ["new cohort reaches 45% weight (0.9x its 50% hash share)"] + [fmt_day(first_day(results[f][0], lambda v: v >= 0.45)) for f in floors], ["new cohort reaches 49% weight"] + [fmt_day(first_day(results[f][0], lambda v: v >= 0.49)) for f in floors], ["last day old cohort holds 2/3 (can lock alone)"] + [fmt_day(max([t for t, v in results[f][1] if v >= TWO_THIRDS] or [0])) for f in floors], ] out = ["### D. Honest growth shock: network doubles at day 60 (1,000 new keys, same total hashrate as the old 1,000)", "", "New cohort's weight share of total (its hashrate share is 50% from day +1):", "", md_table(headers, rows), "", md_table(["event"] + ["f=%d" % f for f in floors], ev)] return "\n".join(out) def pick_offline(rng, weights, target_frac): """Random subset of keys whose weight sums to about target_frac of total (never over by more than one small key).""" total = weights.sum() target = target_frac * total order = rng.permutation(weights.size) mask = np.zeros(weights.size, dtype=bool) acc = 0.0 for i in order: if acc + weights[i] <= target: mask[i] = True acc += weights[i] return mask, acc / total def scenario_e(floors, seed, growth, window, committee, fracs=(0.30, 0.35, 0.50), report_days=35): pick = [1, 2, 3, 5, 7, 10, 15, 20, 25, 30, 31] out = [] for frac in fracs: label = "%d%% of weight goes offline permanently at day 60" % round(100 * frac) if frac != 0.30: label += " (supplementary run)" results = {} for f in floors: rng = np.random.default_rng(seed) p = Params(f, growth, window, committee) net = warm_network(p, rng) w = net.weight() off, actual_frac = pick_offline(rng, w, frac) net.online[off] = False net.group[off] = 9 # dead keys s_all, s_active = [], [] for t in range(1, report_days + 1): net.step() s_all.append((t, net.group_weight_share(0))) s_active.append((t, net.group_weight_share(0, total_mask=net.active_mask()))) results[f] = (s_all, s_active, actual_frac, int(off.sum())) rows = [] for t in pick: rows.append(["+%d" % t] + [fmt_pct(dict(results[f][0])[t]) + " / " + fmt_pct(dict(results[f][1])[t]) for f in floors]) headers = ["day after churn"] + ["f=%d (all keys / active keys)" % f for f in floors] ev = [ ["offline fraction actually picked (keys)"] + ["%s (%d keys)" % (fmt_pct(results[f][2]), results[f][3]) for f in floors], ["live share >= 2/3, total = all keys, first day"] + [fmt_day(first_day(results[f][0], lambda v: v >= TWO_THIRDS)) for f in floors], ["live share >= 2/3, total = active keys, first day"] + [fmt_day(first_day(results[f][1], lambda v: v >= TWO_THIRDS)) for f in floors], ["dead weight fully out of the window (day)"] + [str(window) for _ in floors], ["live share, all-keys total, day +%d" % window] + [fmt_pct(dict(results[f][0])[window]) for f in floors], ] out.append("### E. Churn: %s" % label) out.append("") out.append("Live honest weight share. Left of the slash: total = every key ever seen. " "Right: total = keys that signed a checkpoint in the last 24 h (committee %d)." % committee) out.append("") out.append(md_table(headers, rows)) out.append("") out.append(md_table(["event"] + ["f=%d" % f for f in floors], ev)) out.append("") return "\n".join(out) def scenario_f(floors, seed, growth, window, committee, owner_shares=(0.51, 0.67), days=90): out = [] pick = [15, 30, 45, 60, 90] for s in owner_shares: label = "owner holds %s of hashrate from day 0 and mines honestly" % fmt_pct(s) if s != 0.51: label += " (supplementary run)" results = {} for f in floors: rng = np.random.default_rng(seed) p = Params(f, growth, window, committee) net = Network(p, rng) net.add_keys(pareto_hashrates(rng, N_HONEST, total=1.0 - s), group=0) net.add_keys(np.array([s]), group=1) series = [] for t in range(1, days + 1): net.step() series.append((t, net.group_weight_share(1))) results[f] = series rows = [] for t in pick: rows.append([str(t)] + [fmt_pct(dict(results[f])[t]) for f in floors]) tail = lambda f: [v for t, v in results[f] if t > 30] ev = [ ["mean weight share, days 31 to %d" % days] + [fmt_pct(np.mean(tail(f))) for f in floors], ["min / max weight share, days 31 to %d" % days] + ["%s / %s" % (fmt_pct(min(tail(f))), fmt_pct(max(tail(f)))) for f in floors], ["days at or above 2/3 (can lock alone), of %d" % days] + [str(sum(1 for _, v in results[f] if v >= TWO_THIRDS)) for f in floors], ["days above 1/3 (can veto every lock), of %d" % days] + [str(sum(1 for _, v in results[f] if v > 1.0 / 3.0)) for f in floors], ] out.append("### F. Patient owner: %s" % label) out.append("") out.append("Owner's weight share of total by day:") out.append("") out.append(md_table(["day"] + ["f=%d" % f for f in floors], rows)) out.append("") out.append(md_table(["event"] + ["f=%d" % f for f in floors], ev)) out.append("") return "\n".join(out) SCENARIOS = {"A": scenario_a, "B": scenario_b, "C": scenario_c, "D": scenario_d, "E": scenario_e, "F": scenario_f} def main(argv=None): ap = argparse.ArgumentParser(description="Igneum sustained-mining finality vote-weight simulation") ap.add_argument("--seed", type=int, default=7) ap.add_argument("--scenarios", default="A,B,C,D,E,F") ap.add_argument("--floors", default="1,10,100,1000", help="comma list of floor f values (blocks/day)") ap.add_argument("--growth", type=float, default=2.0, help="daily growth cap multiplier (2 = each day at most 2x the day before)") ap.add_argument("--window", type=int, default=30, help="trailing window in days") ap.add_argument("--committee", type=int, default=100, help="checkpoint committee size, used only for the active-total definition in E") ap.add_argument("--presence", type=int, default=0, help="proposed presence gate: a key's daily cap stays at the floor until it has counted blocks on this many of the window days (0 = off, the rule as written)") args = ap.parse_args(argv) Params.default_presence = args.presence floors = [int(x) for x in args.floors.split(",") if x] print("# finality_sim output") print() print("seed %d, growth cap %.1fx, window %d days, committee %d, presence gate %d, floors %s, blocks/day %d, honest keys %d, Pareto shape %.1f" % (args.seed, args.growth, args.window, args.committee, args.presence, floors, BLOCKS_PER_DAY, N_HONEST, PARETO_SHAPE)) print() for s in args.scenarios.split(","): s = s.strip().upper() if s not in SCENARIOS: print("unknown scenario %s" % s, file=sys.stderr) return 2 print(SCENARIOS[s](floors, args.seed, args.growth, args.window, args.committee)) print() return 0 if __name__ == "__main__": sys.exit(main())