#!/usr/bin/env python3 """Potatium Hub/Worker PoP Monte-Carlo simulation. RESEARCH ONLY; NOT an on-chain/protocol implementation. All PTM amounts and probabilities are floats solely to compare hypotheses, never for consensus. No network, no wallet keys, no third-party Python dependencies. Simplifications: independent Bernoulli proof success, identical CPUs, always online, no sybil resistance, single consensus decision per round, no hash-time or chain latency, no network costs. The rational policy uses a *heuristic*. """ from __future__ import annotations import argparse import hashlib import csv import json import math import random import statistics from dataclasses import dataclass, field from pathlib import Path WINDOW_PROFILES = { "balanced": (0.55, 0.30, 0.20), "original": (0.70, 0.70, 0.70), "cautious": (0.50, 0.30, 0.20), } WINDOW_FEE_MULTIPLIERS = {"original": 1.0, "mild": 0.1, "low": 0.01} HISTORY_DAYS = 90 MAX_WORKERS = 10 # simulation boundary, NOT an approved protocol limit def worker_fee(count: int, model: str = "marginal", scale: float = 1.0) -> float: """Total fee in PTM for activating count paid IDs for one PoP round. 'marginal': per-worker marginal fees (1-3 .001, 4-7 .003, 8-10 .005). 'tier': one collective tier fee for an entire operator. """ if not isinstance(count, int) or not 0 <= count <= MAX_WORKERS: raise ValueError("worker count must be an integer in range 0..10") if model not in ("tier", "marginal") or not math.isfinite(scale) or scale < 0: raise ValueError("invalid worker fee model/scale") if model == "marginal": base = (min(count, 3) * .001 + min(max(count - 3, 0), 4) * .003 + max(count - 7, 0) * .005) else: base = 0.0 if count == 0 else (.001 if count <= 3 else .003 if count <= 7 else .005) return base * scale def progressive_fee(rewards_90d: float) -> float: """Exact baseline fee *formula* from v0.2 research, still nonbinding.""" if not math.isfinite(rewards_90d) or rewards_90d < 0: raise ValueError("invalid 90d rewards") return (min(rewards_90d, 10.0) * .00005 + min(max(rewards_90d - 10.0, 0.0), 90.0) * .0002 + max(rewards_90d - 100.0, 0.0) * .0005) def window_fee(window: int, rewards_90d: float, model: str) -> float: if window == 1: return 0.0 if model not in WINDOW_FEE_MULTIPLIERS or window not in (2, 3): raise ValueError("invalid window/model") progressive = progressive_fee(rewards_90d) * WINDOW_FEE_MULTIPLIERS[model] return (.001 + progressive) if window == 2 else (.003 + 2 * progressive) def gini(nonnegative: list[float]) -> float: if any(not math.isfinite(x) or x < -1e-8 for x in nonnegative): raise ValueError("invalid balances") values = sorted(max(0., x) for x in nonnegative) if not values or sum(values) == 0: return 0.0 n = len(values) return 2 * sum((i + 1) * x for i, x in enumerate(values)) / (n * sum(values)) - (n + 1) / n @dataclass class Operator: ident: int target: bool desired_workers: int policy: str join_day: int = 0 balance: float = 0.0 recent: float = 0.0 earned: float = 0.0 burned_worker: float = 0.0 burned_windows: float = 0.0 wins: int = 0 tickets: int = 0 active_workers_total: int = 0 window2: int = 0 window3: int = 0 skipped_worker_budget: int = 0 skipped_window_budget: int = 0 skipped_window_policy: int = 0 history: list[float] = field(default_factory=lambda: [0.0] * HISTORY_DAYS) def expire_day(self, day: int) -> None: idx = day % HISTORY_DAYS self.recent = max(0.0, self.recent - self.history[idx]) self.history[idx] = 0.0 def add_win(self, day: int, amount: float) -> None: self.balance += amount self.earned += amount self.recent += amount self.history[day % HISTORY_DAYS] += amount self.wins += 1 @property def total_fees(self) -> float: return self.burned_worker + self.burned_windows def activate_workers(op: Operator, model: str, scale: float) -> int: """Pay full activation charge BEFORE draws. Cost is incremental by count. A Hub is free. If budget is short, activate the most affordable prefix. The node/miner never receives a wallet private key in this abstract model. """ wanted = op.desired_workers actual = wanted while actual > 0 and worker_fee(actual, model, scale) > op.balance + 1e-10: actual -= 1 charge = worker_fee(actual, model, scale) if actual < wanted: op.skipped_worker_budget += wanted - actual op.balance -= charge op.burned_worker += charge op.active_workers_total += actual return actual def marginal_share_value(reward: float, own_tickets: int, external_tickets: float) -> float: """Heuristic additional operator payout if another of its nodes qualifies. Given k current operator tickets and m expected external tickets, an additional ticket increases its payout share by m/((k+m)(k+m+1)). With k=0, it creates the first operator ticket: 1/(m+1). """ if own_tickets < 0 or external_tickets < 0: raise ValueError("invalid ticket estimate") if own_tickets == 0: return reward / (1 + external_tickets) total = own_tickets + external_tickets return reward * external_tickets / (total * (total + 1)) def should_buy(op: Operator, window: int, charge: float, profile: tuple[float, float, float], window_model: str, reward: float, external_ticket_estimate: float, own_qualified: int) -> bool: if op.balance + 1e-10 < charge: op.skipped_window_budget += 1 return False if op.policy == "free": op.skipped_window_policy += 1 return False if op.policy == "always": return True if op.policy != "rational": raise ValueError("invalid policy") # Simplified, deliberately non-optimal economic rule. No future information. per_ticket_value = marginal_share_value(reward, own_qualified, external_ticket_estimate) if window == 2: future_charge = window_fee(3, op.recent, window_model) future_value = (max(0.0, profile[2] * per_ticket_value - future_charge) if op.balance - charge + 1e-10 >= future_charge else 0.0) expected = profile[1] * per_ticket_value + (1 - profile[1]) * future_value else: expected = profile[2] * per_ticket_value if expected > charge + 1e-12: return True op.skipped_window_policy += 1 return False def simulate(*, operators: int = 20, target_workers: int = 3, others_workers: int = 0, days: int = 180, rounds_per_day: int = 720, profile_name: str = "balanced", window_model: str = "low", worker_model: str = "marginal", worker_scale: float = 1.0, target_policy: str = "rational", others_policy: str = "free", reward: float = 1.0, seed: int = 20261025, newcomers: int = 0, join_day: int = 90, initial_balance: float = 0.0) -> dict: """Simulate one target operating a free Hub plus W paid mining Workers. All nodes independently qualify; exactly one ticket per node maximum. All PTM, mining income, 90d history and worker/window costs aggregate by operator. Other operators are full competitors, not a static ticket-rate assumption. """ if not (1 <= operators <= 1000 and 0 <= newcomers <= 1000 and 1 <= days <= 3650 and 1 <= rounds_per_day <= 100000 and 0 <= join_day and isinstance(seed, int) and math.isfinite(reward) and reward > 0 and math.isfinite(initial_balance) and initial_balance >= 0): raise ValueError("invalid simulation size, reward, seed, or initial balance") if (not isinstance(target_workers, int) or not 0 <= target_workers <= MAX_WORKERS or not isinstance(others_workers, int) or not 0 <= others_workers <= MAX_WORKERS): raise ValueError("worker count outside research range 0..10") if profile_name not in WINDOW_PROFILES or window_model not in WINDOW_FEE_MULTIPLIERS: raise ValueError("unknown PoP profile/window fee model") if worker_model not in ("tier", "marginal") or not math.isfinite(worker_scale) or worker_scale < 0: raise ValueError("invalid worker fee model/scale") if target_policy not in ("free", "rational", "always") or others_policy not in ("free", "rational", "always"): raise ValueError("invalid window policy") profile = WINDOW_PROFILES[profile_name] rng = random.Random(seed) persons = [Operator(i, i == 0, target_workers if i == 0 else others_workers, target_policy if i == 0 else others_policy, balance=initial_balance) for i in range(operators)] persons += [Operator(operators + i, False, others_workers, others_policy, join_day=join_day, balance=initial_balance) for i in range(newcomers)] target = persons[0] issued = 0.0 no_ticket = 0 total_tickets = 0 daily = [] for day in range(days): living = [op for op in persons if op.join_day <= day] for op in living: op.expire_day(day) day_worker_burn = day_window_burn = day_wins = day_tickets = 0 for _ in range(rounds_per_day): # First pay worker activation for *all* operators, then release F1. nodes: list[Operator] = [] for op in living: n = activate_workers(op, worker_model, worker_scale) nodes.extend([op] * (1 + n)) # free Hub plus paid worker identities rng.shuffle(nodes) # fair order if fees cannot cover every window qualified: list[Operator] = [] failures: list[Operator] = [] own_tickets: dict[int, int] = {op.ident: 0 for op in living} ids_per_operator: dict[int, int] = {} for op in nodes: ids_per_operator[op.ident] = ids_per_operator.get(op.ident, 0) + 1 for op in nodes: if rng.random() < profile[0]: qualified.append(op) own_tickets[op.ident] += 1 else: failures.append(op) # F2/F3 charging happens strictly before drawing the challenge result. # Estimate unknown rival tickets from their active mining IDs. # This deliberately does not peek at actual rivals' F1 outcomes. # Their paid-window strategy is omitted from this naive estimate. for stage in (2, 3): if not failures: break rng.shuffle(failures) next_failures: list[Operator] = [] for op in failures: fee = window_fee(stage, op.recent, window_model) rivals = len(nodes) - ids_per_operator[op.ident] if not should_buy(op, stage, fee, profile, window_model, reward, rivals * profile[0], own_tickets[op.ident]): continue # leaves PoP round with no ticket op.balance -= fee op.burned_windows += fee if stage == 2: op.window2 += 1 else: op.window3 += 1 if rng.random() < profile[stage - 1]: qualified.append(op) own_tickets[op.ident] += 1 elif stage == 2: next_failures.append(op) failures = next_failures for ticket_owner in qualified: ticket_owner.tickets += 1 total_tickets += len(qualified) day_tickets += len(qualified) if qualified: winner = rng.choice(qualified) winner.add_win(day, reward) issued += reward day_wins += 1 else: no_ticket += 1 burned_worker = sum(op.burned_worker for op in persons) burned_windows = sum(op.burned_windows for op in persons) burned_total = burned_worker + burned_windows previous = daily[-1] if daily else None day_worker_burn = burned_worker - (previous["worker_burn_total"] if previous else 0.0) day_window_burn = burned_windows - (previous["window_burn_total"] if previous else 0.0) balances = [op.balance for op in living] daily.append({ "day": day + 1, "active_operators": len(living), "issued_total": round(issued, 6), "worker_burn_total": round(burned_worker, 6), "window_burn_total": round(burned_windows, 6), "burn_total": round(burned_total, 6), "supply": round(initial_balance * len(persons) + issued - burned_total, 6), "gini_operator_wallets": round(gini(balances), 6), "target_wallet": round(target.balance, 6), "target_90d_wins": round(target.recent, 6), "rounds_won": day_wins, "tickets": day_tickets, "worker_burn_today": round(day_worker_burn, 6), "window_burn_today": round(day_window_burn, 6), }) n_rounds = days * rounds_per_day burned_worker = sum(op.burned_worker for op in persons) burned_windows = sum(op.burned_windows for op in persons) burn = burned_worker + burned_windows final_balances = [op.balance for op in persons] conservation = initial_balance * len(persons) + issued - burn # The daily rounding is presentation only; raw floats are used for checks. assert all(op.balance >= -1e-7 for op in persons) assert math.isclose(sum(op.earned for op in persons), issued, rel_tol=1e-9, abs_tol=1e-5) assert math.isclose(sum(final_balances), conservation, rel_tol=1e-9, abs_tol=1e-5) assert sum(op.wins for op in persons) + no_ticket == n_rounds all_mining_ids = sum(1 + op.desired_workers for op in persons) return { "parameters": { "operators": operators, "newcomers": newcomers, "join_day": join_day, "target_workers": target_workers, "others_workers": others_workers, "days": days, "rounds_per_day": rounds_per_day, "profile": profile_name, "success_per_window": profile, "window_fee_model": window_model, "worker_fee_model": worker_model, "worker_fee_scale": worker_scale, "target_policy": target_policy, "others_policy": others_policy, "reward": reward, "seed": seed, "initial_balance": initial_balance, }, "summary": { "rounds": n_rounds, "no_ticket_rounds": no_ticket, "minted": round(issued, 6), "burned_workers": round(burned_worker, 6), "burned_windows": round(burned_windows, 6), "burned_total": round(burn, 6), "circulating": round(conservation, 6), "burn_percent_of_minted": round(burn / issued * 100, 4) if issued else 0.0, "gini_operator_wallets": round(gini(final_balances), 6), "total_tickets": total_tickets, "target_wins": target.wins, "target_win_percent": round(target.wins / n_rounds * 100, 6), "target_ticket_percent": round(target.tickets / total_tickets * 100, 6) if total_tickets else 0.0, "target_earned": round(target.earned, 6), "target_worker_fees": round(target.burned_worker, 6), "target_window_fees": round(target.burned_windows, 6), "target_net": round(target.earned - target.total_fees, 6), "target_wallet": round(target.balance, 6), "target_recent90": round(target.recent, 6), "target_avg_active_workers": round(target.active_workers_total / n_rounds, 4), "target_window2": target.window2, "target_window3": target.window3, "target_worker_budget_denials": target.skipped_worker_budget, "target_window_budget_denials": target.skipped_window_budget, "unused_possible_id_count": all_mining_ids, }, "operators": [ {"id": op.ident, "target": op.target, "join_day": op.join_day, "workers_requested": op.desired_workers, "policy": op.policy, "wins": op.wins, "tickets": op.tickets, "earned": round(op.earned, 6), "paid_workers": round(op.burned_worker, 6), "paid_windows": round(op.burned_windows, 6), "net": round(op.earned - op.total_fees, 6), "wallet": round(op.balance, 6), "last90": round(op.recent, 6), "avg_active_workers": round(op.active_workers_total / n_rounds, 4), "window2": op.window2, "window3": op.window3, "skipped_worker_budget": op.skipped_worker_budget, "skipped_window_budget": op.skipped_window_budget, "skipped_window_policy": op.skipped_window_policy} for op in persons ], "daily": daily, "warnings": [ "RESEARCH ONLY: float is prohibited in consensus state/fee computation.", "Unlinked owner identities and Sybil attacks not modeled; operator registry assumed valid.", "Mining proofs are independent Bernoulli outcomes, not real bounded-hash attempts.", "Windows simulated sequentially per PoP round; timing/network/validator censor absent.", "Worker activation before F1 is charged even if no worker qualifies.", "Rational is a naive per-operator marginal-payout heuristic, not a validated optimal strategy.", "Rational expects rivals' F1 success only, not their paid-window decisions or future actions.", "No multi-instance-for-one-physical-device detection or wallet transfer attacks.", "Concentration/Gini refers to registered operators, not real persons.", "Rewards and block count/round length are illustrative, not protocol parameters.", ], } def compare(scenarios: list[dict]) -> list[dict]: """Return scenario comparison and marginal net gains per additional worker. Comparing requested W is not a counterfactual with identical RNG paths. Pair many seeds to reduce noise. Marginal gain divided by count difference. """ rows = [] ordered = sorted(scenarios, key=lambda r: (r["parameters"]["target_workers"], r["parameters"]["seed"])) workers = sorted({r["parameters"]["target_workers"] for r in scenarios}) for w in workers: ss = [r["summary"] for r in ordered if r["parameters"]["target_workers"] == w] if not ss: continue sample = { "workers": w, "replicates": len(ss), "win_pct": statistics.mean(s["target_win_percent"] for s in ss), "avg_active": statistics.mean(s["target_avg_active_workers"] for s in ss), "net_ptm": statistics.mean(s["target_net"] for s in ss), "gross_ptm": statistics.mean(s["target_earned"] for s in ss), "worker_burn": statistics.mean(s["target_worker_fees"] for s in ss), "window_burn": statistics.mean(s["target_window_fees"] for s in ss), "burn_pct": statistics.mean(s["burn_percent_of_minted"] for s in ss), "gini": statistics.mean(s["gini_operator_wallets"] for s in ss), "wallet": statistics.mean(s["target_wallet"] for s in ss), "window2": statistics.mean(s["target_window2"] for s in ss), "window3": statistics.mean(s["target_window3"] for s in ss), } sample["net_range"] = (min(s["target_net"] for s in ss), max(s["target_net"] for s in ss)) rows.append(sample) for i, row in enumerate(rows): row["delta_net_per_added_worker"] = None if not i else ( (row["net_ptm"] - rows[i-1]["net_ptm"]) / (row["workers"]- rows[i-1]["workers"])) base = next((r["net_ptm"] for r in rows if r["workers"] == 0), None) for row in rows: row["net_gain_vs_hub"] = None if base is None else row["net_ptm"] - base return rows def write_results(out: Path, scenarios: list[dict], rows: list[dict], tag: str) -> list[Path]: out.mkdir(parents=True, exist_ok=True) stem = "pop_cluster_" + tag paths = [out / (stem + ext) for ext in (".txt", ".json", "_comparison.csv", "_daily.csv", "_operators.csv")] t, j, c, d, o = paths j.write_text(json.dumps({"comparison": rows, "scenarios": scenarios}, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") with c.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=[key for key in rows[0] if key != "net_range"]) writer.writeheader() writer.writerows({k: val for k, val in row.items() if k != "net_range"} for row in rows) with d.open("w", newline="", encoding="utf-8") as handle: fields = ["workers", "seed"] + list(scenarios[0]["daily"][0].keys()) writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for r in scenarios: for day in r["daily"]: writer.writerow({"workers": r["parameters"]["target_workers"], "seed": r["parameters"]["seed"], **day}) with o.open("w", newline="", encoding="utf-8") as handle: fields = ["scenario_workers", "seed"] + list(scenarios[0]["operators"][0].keys()) writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for r in scenarios: for op in r["operators"]: writer.writerow({"scenario_workers": r["parameters"]["target_workers"], "seed": r["parameters"]["seed"], **op}) p = scenarios[0]["parameters"] lines = [ "POTATIUM | Hub + Worker + 3 Fenster | FORSCHUNG, KEINE PROTOKOLLREGEL", f"Operatoren={p['operators']} (+{p['newcomers']} ab Tag {p['join_day']}) | Tage={p['days']} | Runden/Tag={p['rounds_per_day']}", f"PoP={p['profile']} {p['success_per_window']} | Fenster={p['window_fee_model']} | Worker={p['worker_fee_model']} x{p['worker_fee_scale']:g}", f"Strategie Ziel={p['target_policy']} Andere={p['others_policy']} (je {p['others_workers']} Worker) | Reward={p['reward']:g} PTM", f"Seeds: {', '.join(str(s) for s in sorted({r['parameters']['seed'] for r in scenarios}))}", "", " W Ø aktive Gewinn% Brutto PTM Netto PTM Worker-Geb. Fenster-Geb. ΔNetto/Worker Gini", ] for r in rows: delta = " n/a" if r["delta_net_per_added_worker"] is None else f"{r['delta_net_per_added_worker']:>13.4f}" lines.append(f"{r['workers']:>2} {r['avg_active']:>9.2f} {r['win_pct']:>8.3f} " f"{r['gross_ptm']:>11.2f} {r['net_ptm']:>11.2f} " f"{r['worker_burn']:>12.2f} {r['window_burn']:>12.2f} {delta:>14} {r['gini']:>5.3f}") lines.extend(["", "WARNUNGEN:"] + ["- " + w for w in scenarios[0]["warnings"]]) lines.append("- Monte-Carlo-Rauschen: Δ Netto ist bei wenigen Seeds KEINE sichere Grenzkosten-Schwelle.") t.write_text("\n".join(lines) + "\n", encoding="utf-8") print("\n".join(lines)) return paths def main() -> None: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument("--operators", type=int, default=20, help="one target + rest competing operators") parser.add_argument("--workers", type=int, nargs="+", default=[0, 1, 2, 3, 5, 8, 10], help="requested target Workers to compare, Hub is additionally free") parser.add_argument("--others-workers", type=int, default=0) parser.add_argument("--days", type=int, default=180) parser.add_argument("--rounds-per-day", type=int, default=720) parser.add_argument("--seed", type=int, default=20261025) parser.add_argument("--replicates", type=int, default=1, help="number of seeds, e.g. 3") parser.add_argument("--profile", choices=WINDOW_PROFILES, default="balanced") parser.add_argument("--window-fee-model", choices=WINDOW_FEE_MULTIPLIERS, default="low") parser.add_argument("--worker-fee-model", choices=["tier", "marginal"], default="marginal") parser.add_argument("--worker-fee-scale", type=float, default=1.0) parser.add_argument("--target-policy", choices=["free", "rational", "always"], default="rational") parser.add_argument("--others-policy", choices=["free", "rational", "always"], default="free") parser.add_argument("--newcomers", type=int, default=0) parser.add_argument("--join-day", type=int, default=90) parser.add_argument("--initial-balance", type=float, default=0.0) parser.add_argument("--reward", type=float, default=1.0) parser.add_argument("--out", type=Path, default=Path.home() / "potatium-simulation") a = parser.parse_args() if a.replicates < 1 or a.replicates > 100 or not a.workers or len(set(a.workers)) != len(a.workers): parser.error("replicates must be 1..100 and --workers must contain unique values") if a.operators < 2: parser.error("require at least two competing operators") try: # Catch invalid arguments before running the potentially long sweep. worker_fee(max(a.workers), a.worker_fee_model, a.worker_fee_scale) if min(a.workers) < 0: raise ValueError("worker count negative") worker_fee(a.others_workers, a.worker_fee_model, a.worker_fee_scale) if a.newcomers and a.join_day >= a.days: raise ValueError("newcomers must join within simulation horizon") if a.operators + a.newcomers > 300: raise ValueError("max 300 operators in this research script") configs = [] for worker_count in sorted(a.workers): for ix in range(a.replicates): configs.append((worker_count, a.seed + ix)) scenarios = [] for index, (w, seed) in enumerate(configs, 1): print(f"[{index}/{len(configs)}] Worker={w}, Seed={seed} ...", flush=True) scenarios.append(simulate(operators=a.operators, target_workers=w, others_workers=a.others_workers, days=a.days, rounds_per_day=a.rounds_per_day, profile_name=a.profile, window_model=a.window_fee_model, worker_model=a.worker_fee_model, worker_scale=a.worker_fee_scale, target_policy=a.target_policy, others_policy=a.others_policy, reward=a.reward, seed=seed, newcomers=a.newcomers, join_day=a.join_day, initial_balance=a.initial_balance)) rows = compare(scenarios) fingerprint_parameters = { "operators": a.operators, "workers": sorted(a.workers), "others_workers": a.others_workers, "days": a.days, "rounds_per_day": a.rounds_per_day, "seed": a.seed, "replicates": a.replicates, "profile": a.profile, "window_fee_model": a.window_fee_model, "worker_fee_model": a.worker_fee_model, "worker_fee_scale": a.worker_fee_scale, "target_policy": a.target_policy, "others_policy": a.others_policy, "newcomers": a.newcomers, "join_day": a.join_day, "initial_balance": a.initial_balance, "reward": a.reward, } suffix = hashlib.sha256(json.dumps(fingerprint_parameters, sort_keys=True).encode()).hexdigest()[:10] tag = (f"{a.operators}o_{a.days}d_{a.profile}_{a.window_fee_model}_" f"{a.worker_fee_model}_{a.target_policy}_{a.seed}_{suffix}") paths = write_results(a.out, scenarios, rows, tag) except (ValueError, OverflowError) as error: parser.error(str(error)) print("\nErgebnisdateien:") for path in paths: print(" " + str(path)) print("\nBitte TXT und Vergleichs-CSV hier hochladen; fuer Tiefenanalyse auch Tages-CSV.") if __name__ == "__main__": main()