#!/usr/bin/env python3 """Potatium: dynamic worker activation, three PoP windows, treasury circulation. Pure Monte-Carlo RESEARCH. No on-chain code, cryptographic PoW, personal keys, commercial price assumptions, or final protocol parameters. Uses only Python standard library; floats are NOT acceptable for consensus. """ from __future__ import annotations import argparse import csv import hashlib import json import math import random import statistics import zipfile from dataclasses import dataclass, field from pathlib import Path F1_F2_F3 = {"balanced": (.55, .30, .20), "cautious": (.50, .30, .20), "original": (.70, .70, .70)} WINDOW_SCALES = {"low": .01, "mild": .1, "original": 1.} MAX_WORKERS = 10 # Research range, NOT a protocol cap HISTORY_DAYS = 90 EPS = 1e-9 def progressive_fee(rewards_90d: float) -> float: if not math.isfinite(rewards_90d) or rewards_90d < 0: raise ValueError("invalid rewards_90d") return (min(rewards_90d, 10) * .00005 + min(max(rewards_90d - 10, 0), 90) * .0002 + max(rewards_90d - 100, 0) * .0005) def window_fee(stage: int, rewards_90d: float, model: str = "low") -> float: if stage not in (2, 3) or model not in WINDOW_SCALES: raise ValueError("invalid PoP stage or fee model") p = progressive_fee(rewards_90d) * WINDOW_SCALES[model] return (.001 + p) if stage == 2 else (.003 + 2 * p) def worker_fee(count: int, scale: float) -> float: if not isinstance(count, int) or not 0 <= count <= MAX_WORKERS or not math.isfinite(scale) or scale < 0: raise ValueError("invalid worker count or fee scale") base = min(count, 3) * .001 + min(max(count - 3, 0), 4) * .003 + max(count - 7, 0) * .005 return scale * base def gini(balances: list[float]) -> float: if any(not math.isfinite(x) or x < -EPS for x in balances): raise ValueError("invalid balance") sorted_bal = sorted(max(0., b) for b in balances) total = sum(sorted_bal) if total <= 0: return 0. return (2 * sum((i+1)*v for i,v in enumerate(sorted_bal)) / (len(sorted_bal)*total) - (len(sorted_bal)+1)/len(sorted_bal)) @dataclass class Operator: id: int capacity: int join_day: int balance: float recent: float = 0. earned: float = 0. worker_paid: float = 0. window_paid: float = 0. rebated: float = 0. wins: int = 0 tickets: int = 0 activation_total: int = 0 activation_rounds: int = 0 paid_f2: int = 0 paid_f3: int = 0 refused_budget: int = 0 history: list[float] = field(default_factory=lambda: [0.]*HISTORY_DAYS) def expire(self, day: int) -> None: idx = day % HISTORY_DAYS self.recent -= self.history[idx] self.history[idx] = 0. if abs(self.recent) < EPS: self.recent = 0. def reward(self, day: int, amount: float) -> None: self.earned += amount self.balance += amount self.wins += 1 self.history[day % HISTORY_DAYS] += amount self.recent += amount @property def fees(self) -> float: return self.worker_paid + self.window_paid def marginal_ticket_gain(own: int, rival_est: float, reward: float) -> float: if own < 0 or rival_est < 0: raise ValueError("ticket forecast invalid") if own == 0: return reward / (1 + rival_est) total = own + rival_est return reward * rival_est / (total * (total + 1)) def choose_worker_count(*, balance: float, capacity: int, mode: str, scale: float, rivals_estimated_tickets: float, f1_chance: float, reward: float, worker_choice_margin: float = 1.0) -> int: """Heuristic worker choice BEFORE current mining proof/challenge. 'fixed' requests all capacity (budget may reduce it). 'dynamic' maximizes [estimated ticket share * reward - worker fee] against all affordable capacities. Only F1 chance informs forecast, hence a deliberately imperfect/myopic policy; it cannot observe live rival proofs. """ if mode not in ("fixed", "dynamic") or capacity < 0 or capacity > MAX_WORKERS: raise ValueError("invalid worker choice") if balance < -EPS or not 0 < f1_chance <= 1 or reward <= 0 or worker_choice_margin <= 0: raise ValueError("invalid worker choice parameter") affordable = [n for n in range(capacity+1) if worker_fee(n, scale) <= balance+EPS] if mode == "fixed": return affordable[-1] def net(n: int) -> float: own = (n + 1) * f1_chance return worker_choice_margin * reward * own / (own + max(rivals_estimated_tickets, EPS)) - worker_fee(n, scale) # tie-break toward fewer workers return max(affordable, key=lambda n: (net(n), -n)) def should_buy_window(*, policy: str, op_balance: float, charge: float, stage: int, own_tickets: int, rival_estimate: float, reward: float, probabilities: tuple[float,float,float], future_charge: float = 0.) -> bool: if op_balance + EPS < charge or policy == "free": return False if policy == "always": return True if policy != "rational": raise ValueError("unknown window policy") value = marginal_ticket_gain(own_tickets, rival_estimate, reward) if stage == 2: future = max(0., probabilities[2] * value - future_charge) if op_balance + EPS >= charge + future_charge else 0. expected = probabilities[1] * value + (1 - probabilities[1]) * future elif stage == 3: expected = probabilities[2] * value else: raise ValueError("invalid stage") return expected > charge + 1e-12 def normalize_capacities(operator_count: int, distribution: list[int], newcomers: int) -> list[int]: if not distribution or any(not isinstance(v, int) or v < 0 or v > MAX_WORKERS for v in distribution): raise ValueError("worker distribution requires integers 0..10") return [distribution[i % len(distribution)] for i in range(operator_count + newcomers)] def simulate(*, operators: int = 20, newcomers: int = 5, join_day: int = 90, capacities: tuple[int,...] = (0,1,3,5,8,10), days: int = 180, rounds_per_day: int = 72, seed: int = 20261025, worker_mode: str = "dynamic", worker_scale: float = 3., fund_mode: str = "hold", payout_share: float = .25, payout_interval: int = 30, window_policy: str = "rational", window_model: str = "low", profile: str = "balanced", reward: float = 1., initial_balance: float = 0.) -> dict: """Run a full PoP lottery with conservation-checked on-chain-equivalent fund ledger. Genesis allocations initial_balance are allocated to *all* operators, including future joiners, from day 0; no external minting upon joining. Fund refunds use equal active-operator allocation, not an endorsed policy. """ if not (2 <= operators <= 300 and 0 <= newcomers <= 300 and operators + newcomers <= 300 and days >= 1 and rounds_per_day >= 1 and 0 <= join_day and (newcomers == 0 or join_day < days) and isinstance(seed, int) and math.isfinite(initial_balance) and initial_balance >= 0 and math.isfinite(reward) and reward > 0 and worker_mode in ("fixed", "dynamic") and fund_mode in ("hold", "redistribute") and window_policy in ("free", "rational", "always") and profile in F1_F2_F3 and window_model in WINDOW_SCALES and 0 <= payout_share <= 1 and payout_interval >= 1): raise ValueError("invalid simulation parameters") # Check every capacity and scale early. caps = normalize_capacities(operators, list(capacities), newcomers) worker_fee(max(caps), worker_scale) probabilities = F1_F2_F3[profile] rng = random.Random(seed) persons = [Operator(i, caps[i], 0 if i < operators else join_day, initial_balance) for i in range(operators + newcomers)] fund = 0. fund_in = fund_out = minted = 0. no_ticket_rounds = 0 total_tickets = 0 daily = [] # Prior-round info only. Default forecast: all rivals' hubs qualify at F1. last_rival_tickets = {op.id: (operators-1) * probabilities[0] for op in persons} for day in range(days): living = [op for op in persons if op.join_day <= day] for op in living: op.expire(day) day_wins = 0 day_activations = 0 for _ in range(rounds_per_day): nodes: list[Operator] = [] current_counts: dict[int,int] = {} # All operator decisions occur before F1 challenge, no peeking at outcomes. for op in living: chosen = choose_worker_count( balance=op.balance, capacity=op.capacity, mode=worker_mode, scale=worker_scale, rivals_estimated_tickets=last_rival_tickets[op.id], f1_chance=probabilities[0], reward=reward) if chosen < op.capacity: op.refused_budget += int(worker_fee(chosen+1, worker_scale) > op.balance + EPS) fee = worker_fee(chosen, worker_scale) if fee > 0: op.balance -= fee op.worker_paid += fee fund += fee fund_in += fee op.activation_total += chosen op.activation_rounds += 1 day_activations += chosen current_counts[op.id] = chosen + 1 nodes.extend([op]*(chosen + 1)) # Each node ID is represented exactly once per round. rng.shuffle(nodes) qualified: list[Operator] = [] failures: list[Operator] = [] tickets_by_owner = {op.id: 0 for op in living} for node_owner in nodes: if rng.random() < probabilities[0]: qualified.append(node_owner) tickets_by_owner[node_owner.id] += 1 else: failures.append(node_owner) for stage in (2,3): if not failures: break rng.shuffle(failures) again = [] # Current rival outcomes remain unknown, only previous-round forecast used. for op in failures: cost = window_fee(stage, op.recent, window_model) prospective_f3 = window_fee(3, op.recent, window_model) if stage == 2 else 0. if not should_buy_window( policy=window_policy, op_balance=op.balance, charge=cost, stage=stage, own_tickets=tickets_by_owner[op.id], rival_estimate=last_rival_tickets[op.id], reward=reward, probabilities=probabilities, future_charge=prospective_f3): continue op.balance -= cost op.window_paid += cost fund += cost fund_in += cost if stage == 2: op.paid_f2 += 1 else: op.paid_f3 += 1 if rng.random() < probabilities[stage-1]: qualified.append(op) tickets_by_owner[op.id] += 1 elif stage == 2: again.append(op) failures = again for op in qualified: op.tickets += 1 total_tickets += len(qualified) if qualified: winner = rng.choice(qualified) winner.reward(day, reward) minted += reward day_wins += 1 else: no_ticket_rounds += 1 # Update forecasts solely from number of *actual tickets in past round*. total = len(qualified) for op in living: rivals_now = total - tickets_by_owner[op.id] last_rival_tickets[op.id] = .95*last_rival_tickets[op.id] + .05*rivals_now payout_today = 0. if fund_mode == "redistribute" and (day+1) % payout_interval == 0: payout_today = fund * payout_share payout_per_operator = payout_today / len(living) for op in living: op.balance += payout_per_operator op.rebated += payout_per_operator fund -= payout_today fund_out += payout_today balances = [op.balance for op in living] sum_wallets = sum(op.balance for op in persons) expected_supply = initial_balance*len(persons) + minted if not math.isclose(sum_wallets + fund, expected_supply, rel_tol=1e-9, abs_tol=1e-5): raise AssertionError("SUPPLY CONSERVATION FAILED") if any(op.balance < -EPS for op in persons) or fund < -EPS: raise AssertionError("negative wallet or fund") daily.append({ "day": day+1, "active_operators": len(living), "minted": round(minted, 6), "fund": round(fund, 6), "fund_in": round(fund_in, 6), "fund_out": round(fund_out, 6), "fund_payout_today": round(payout_today, 6), "wallet_total": round(sum_wallets, 6), "fund_percent_of_supply": round(100*fund/expected_supply, 5) if expected_supply else 0., "gini_active_wallets": round(gini(balances), 6), "active_workers_per_round": round(day_activations / rounds_per_day, 4), "wins_today": day_wins, }) nrounds = days*rounds_per_day fund_paid_worker = sum(op.worker_paid for op in persons) fund_paid_window = sum(op.window_paid for op in persons) assert sum(op.wins for op in persons) + no_ticket_rounds == nrounds assert math.isclose(fund_paid_worker+fund_paid_window, fund_in, rel_tol=1e-8, abs_tol=1e-5) assert math.isclose(fund_in - fund_out, fund, rel_tol=1e-8, abs_tol=1e-5) assert math.isclose(sum(op.rebated for op in persons), fund_out, rel_tol=1e-8, abs_tol=1e-5) assert math.isclose(sum(op.earned for op in persons), minted, rel_tol=1e-8, abs_tol=1e-5) supply = initial_balance*len(persons) + minted rows = [] for op in persons: active_rounds = max(0, days-op.join_day) * rounds_per_day rows.append({ "id": op.id, "capacity": op.capacity, "join_day": op.join_day, "active_rounds": active_rounds, "avg_workers_active": round(op.activation_total/active_rounds, 5) if active_rounds else 0., "wins": op.wins, "tickets": op.tickets, "earned": round(op.earned, 5), "paid_worker": round(op.worker_paid, 5), "paid_window": round(op.window_paid, 5), "fund_received": round(op.rebated, 5), "mining_net": round(op.earned-op.fees, 5), "wallet": round(op.balance, 5), "paid_f2": op.paid_f2, "paid_f3": op.paid_f3, "budget_restrictions": op.refused_budget, "recent90": round(op.recent, 5) }) return { "params": {"operators": operators, "newcomers": newcomers, "join_day": join_day, "capacities": list(capacities), "days": days, "rounds_per_day": rounds_per_day, "seed": seed, "worker_mode": worker_mode, "worker_scale": worker_scale, "fund_mode": fund_mode, "payout_share": payout_share, "payout_interval": payout_interval, "window_policy": window_policy, "window_model": window_model, "profile": profile, "probabilities": list(probabilities), "reward": reward, "initial_balance": initial_balance}, "summary": { "rounds": nrounds, "minted": round(minted, 5), "genesis_supply": initial_balance*len(persons), "total_supply": round(supply, 5), "burned": 0., "fund_balance": round(fund, 5), "fund_received_fees": round(fund_in, 5), "fund_distributed": round(fund_out, 5), "fund_paid_worker": round(fund_paid_worker, 5), "fund_paid_window": round(fund_paid_window, 5), "fund_pct_supply": round(100*fund/supply, 6) if supply else 0., "total_wallets": round(sum(op.balance for op in persons), 5), "gini_operator_wallets": round(gini([op.balance for op in persons]), 6), "no_ticket_rounds": no_ticket_rounds, "ticket_count": total_tickets, "total_avg_workers": round(sum(op.activation_total for op in persons)/nrounds, 4), "total_wins_newcomers": sum(op.wins for op in persons if op.join_day > 0), "total_fund_to_newcomers": round(sum(op.rebated for op in persons if op.join_day > 0), 5), "total_fund_to_old": round(sum(op.rebated for op in persons if op.join_day == 0), 5), }, "operators": rows, "daily": daily, "limitations": [ "RESEARCH ONLY: no PoW, signatures, block times, network outages, adversarial miners, exchange prices or actual hardware.", "One operator per registered identity is assumed; no Sybil resistance or proof of physical hardware.", "All operators have one free Hub; capacities are the number of additional paid Workers, not extra free Hubs.", "Dynamic workers use an imperfect myopic F1-only expected-share estimate and a past-round EMA of rival tickets.", "Window strategy is an approximation, not an optimal game-theoretic solution.", "Redistribute allocates every recipient an EQUAL share, including the miner, and is NOT proposed governance or treasury policy.", "Future joiners receive their configurable genesis balance at genesis; only begin competing and receiving payouts upon joining.", "Fund is an internal ledger with conservation checks; no real coins or wallet keys, and no consensus code.", "72 or 720 example mining rounds/day and 1 PTM per winning round are not decided protocol parameters.", "Float amounts are valid for research only, never consensus.", ]} def aggregate(results: list[dict]) -> list[dict]: grouped = {} for r in results: p = r["params"] key = (p["worker_scale"], p["worker_mode"], p["fund_mode"]) grouped.setdefault(key, []).append(r) table = [] for (scale, worker_mode, fund_mode), runs in sorted(grouped.items()): def mean(key: str) -> float: return statistics.mean(r["summary"][key] for r in runs) rows = [op for r in runs for op in r["operators"]] cap_stats = {} for cap in sorted({op["capacity"] for op in rows}): records = [op for op in rows if op["capacity"] == cap and op["join_day"] == 0] if records: cap_stats[f"capacity_{cap}_net"] = round(statistics.mean(op["mining_net"] for op in records),4) cap_stats[f"capacity_{cap}_wallet"] = round(statistics.mean(op["wallet"] for op in records),4) cap_stats[f"capacity_{cap}_gross"] = round(statistics.mean(op["earned"] for op in records),4) cap_stats[f"capacity_{cap}_rebate"] = round(statistics.mean(op["fund_received"] for op in records),4) cap_stats[f"capacity_{cap}_activated"] = round(statistics.mean(op["avg_workers_active"] for op in records),4) newcomer_rows = [op for op in rows if op["join_day"] > 0] incumbent_rows = [op for op in rows if op["join_day"] == 0] def net_per_active_day(records: list[dict]) -> float: return statistics.mean(op["mining_net"] / (op["active_rounds"] / runs[0]["params"]["rounds_per_day"]) for op in records if op["active_rounds"] > 0) if records else 0. summary = {"worker_scale": scale, "worker_mode": worker_mode, "fund_mode": fund_mode, "seeds": len(runs), "fund_balance": round(mean("fund_balance"), 4), "fund_pct_supply": round(mean("fund_pct_supply"), 4), "fund_distributed": round(mean("fund_distributed"), 4), "fund_worker_in": round(mean("fund_paid_worker"), 4), "fund_window_in": round(mean("fund_paid_window"), 4), "minted": round(mean("minted"), 4), "gini_wallets": round(mean("gini_operator_wallets"), 5), "avg_workers_total": round(mean("total_avg_workers"), 4), "newcomer_wins": round(mean("total_wins_newcomers"), 3), "newcomer_net_daily": round(net_per_active_day(newcomer_rows), 4), "incumbent_net_daily": round(net_per_active_day(incumbent_rows), 4)} summary.update(cap_stats) table.append(summary) return table def write_output(folder: Path, results: list[dict], table: list[dict], name: str) -> list[Path]: folder.mkdir(parents=True, exist_ok=True) files = [folder / (name + x) for x in ("_summary.txt", "_comparison.csv", "_operators.csv", "_daily.csv", "_results.json")] txt, comp, ops, days, js = files comp_cols = list(dict.fromkeys(key for r in table for key in r)) with comp.open("w", newline="", encoding="utf-8") as fh: wr = csv.DictWriter(fh, fieldnames=comp_cols) wr.writeheader() wr.writerows(table) with ops.open("w", newline="", encoding="utf-8") as fh: cols = ["worker_scale", "worker_mode", "fund_mode", "seed"] + list(results[0]["operators"][0]) wr = csv.DictWriter(fh, fieldnames=cols) wr.writeheader() for r in results: for op in r["operators"]: wr.writerow({"worker_scale": r["params"]["worker_scale"], "worker_mode": r["params"]["worker_mode"], "fund_mode": r["params"]["fund_mode"], "seed": r["params"]["seed"], **op}) with days.open("w", newline="", encoding="utf-8") as fh: cols = ["worker_scale", "worker_mode", "fund_mode", "seed"] + list(results[0]["daily"][0]) wr = csv.DictWriter(fh, fieldnames=cols) wr.writeheader() for r in results: for d in r["daily"]: wr.writerow({"worker_scale": r["params"]["worker_scale"], "worker_mode": r["params"]["worker_mode"], "fund_mode": r["params"]["fund_mode"], "seed": r["params"]["seed"], **d}) js.write_text(json.dumps({"scenario_count": len(results), "comparisons": table, "runs": results}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") header = ["POTATIUM | Dynamische Worker / Gebühren / Fonds | FORSCHUNG, KEINE PROTOKOLLREGEL", f"Simulationen: {len(results)} | Seeds pro Konfiguration: {table[0]['seeds']}", "Worker-Fee-Skala Modus Fonds Ø Fonds-PTM Ø Fonds-% Ø Ausgaben Gini Ø aktive Worker"] for r in table: header.append(f"{r['worker_scale']:>6}x {r['worker_mode']:<8} {r['fund_mode']:<12} " f"{r['fund_balance']:>10.2f} {r['fund_pct_supply']:>7.2f}% " f"{r['fund_distributed']:>11.2f} {r['gini_wallets']:>5.3f} " f"{r['avg_workers_total']:>10.2f}") header += ["", "WARNUNGEN:"] + ["- " + s for s in results[0]["limitations"]] txt.write_text("\n".join(header)+"\n", encoding="utf-8") print("\n".join(header)) return files def main() -> None: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter) ap.add_argument("--operators", type=int, default=20) ap.add_argument("--newcomers", type=int, default=5) ap.add_argument("--join-day", type=int, default=90) ap.add_argument("--capacities", nargs="+", type=int, default=[0,1,3,5,8,10], help="repeating operator worker-capacity distribution") ap.add_argument("--days", type=int, default=180) ap.add_argument("--rounds-per-day", type=int, default=72, help="720 for full earlier research cadence; 72 for quick study") ap.add_argument("--seed", type=int, default=20261025) ap.add_argument("--replicates", type=int, default=3) ap.add_argument("--scales", nargs="+", type=float, default=[1.,3.,5.]) ap.add_argument("--worker-modes", nargs="+", choices=["fixed", "dynamic"], default=["fixed", "dynamic"]) ap.add_argument("--fund-modes", nargs="+", choices=["hold", "redistribute"], default=["hold", "redistribute"]) ap.add_argument("--payout-share", type=float, default=.25) ap.add_argument("--payout-interval", type=int, default=30) ap.add_argument("--window-policy", choices=["free", "rational", "always"], default="rational") ap.add_argument("--window-fee-model", choices=WINDOW_SCALES, default="low") ap.add_argument("--profile", choices=F1_F2_F3, default="balanced") ap.add_argument("--reward", type=float, default=1.) ap.add_argument("--initial-balance", type=float, default=0.) ap.add_argument("--out", type=Path, default=Path.home()/"potatium-dynamic-simulation") args = ap.parse_args() if not (1 <= args.replicates <= 30 and args.scales and len(set(args.scales)) == len(args.scales) and 1 <= len(args.scales) <= 10 and len(args.worker_modes) == len(set(args.worker_modes)) and len(args.fund_modes) == len(set(args.fund_modes))): ap.error("replicates 1..30; unique nonempty scales/modes") if any(not math.isfinite(s) or s < 0 for s in args.scales): ap.error("scales must be finite and nonnegative") if args.newcomers and args.join_day >= args.days: ap.error("newcomers must arrive within simulation days") if not 0 <= args.payout_share <= 1 or args.payout_interval <= 0: ap.error("invalid payout configuration") if any(c < 0 or c > MAX_WORKERS for c in args.capacities): ap.error("capacities outside 0..10") results = [] configs = [(scale, mode, fund, args.seed+i) for scale in args.scales for mode in args.worker_modes for fund in args.fund_modes for i in range(args.replicates)] for i, (scale, mode, fund, seed) in enumerate(configs, 1): print(f"[{i}/{len(configs)}] workers={mode}, fee={scale}x, fund={fund}, seed={seed}", flush=True) results.append(simulate( operators=args.operators, newcomers=args.newcomers, join_day=args.join_day, capacities=tuple(args.capacities), days=args.days, rounds_per_day=args.rounds_per_day, seed=seed, worker_mode=mode, worker_scale=scale, fund_mode=fund, payout_share=args.payout_share, payout_interval=args.payout_interval, window_policy=args.window_policy, window_model=args.window_fee_model, profile=args.profile, reward=args.reward, initial_balance=args.initial_balance)) comparisons = aggregate(results) tag_dict = {"operators": args.operators,"newcomers":args.newcomers,"join_day":args.join_day, "capacities":args.capacities,"days":args.days,"rounds":args.rounds_per_day, "seed":args.seed,"replicates":args.replicates,"scales":args.scales, "worker_modes":args.worker_modes,"fund_modes":args.fund_modes, "payout_share":args.payout_share,"payout_interval":args.payout_interval, "policy":args.window_policy,"window_fee":args.window_fee_model, "profile":args.profile,"initial_balance":args.initial_balance,"reward":args.reward} tag = hashlib.sha256(json.dumps(tag_dict,sort_keys=True).encode()).hexdigest()[:10] stem = f"potatium_dynamic_{args.operators}o_{args.days}d_{args.rounds_per_day}r_{tag}" produced = write_output(args.out, results, comparisons, stem) archive = args.out / (stem + "_ERGEBNISSE.zip") with zipfile.ZipFile(archive, "w", zipfile.ZIP_DEFLATED, compresslevel=7) as z: for path in produced: z.write(path, arcname=path.name) with zipfile.ZipFile(archive) as z: if z.testzip() is not None: raise RuntimeError("Ergebnisarchiv ist beschaedigt") print("\nDateien:\n" + "\n".join(str(x) for x in produced)) print("\nZIP fuer Auswertung: " + str(archive)) if __name__ == "__main__": main()