#!/usr/bin/env python3 """Potatium: research-only fee allocation from an existing cluster simulation. No simulation state is modified. This is an ACCOUNTING COUNTERFACTUAL, not an incentive-aware rerun: miners still make the same decisions because fees are paid to recipient accounts outside the participating miner population. Input: existing pop_cluster_*.json produced by simulate_pop_cluster.py, either in a directory or inside a ZIP (including the user's archive). Output: TXT, CSV, JSON and one ZIP for subsequent evaluation. All float calculations are OFF CHAIN and never consensus implementation. """ from __future__ import annotations import argparse import csv import json import math import statistics import zipfile from collections import defaultdict from dataclasses import dataclass from pathlib import Path MAX_FILE_SIZE = 30 * 1024 * 1024 MAX_INPUT_FILES = 100 TOLERANCE = 0.002 # old reports round wallet balances to six decimals @dataclass(frozen=True) class Split: """Fractions for (burn, validator, treasury). Must sum to exactly 1.""" burn: float validators: float treasury: float def __post_init__(self): if any(not math.isfinite(x) or x < 0 or x > 1 for x in (self.burn, self.validators, self.treasury)): raise ValueError('Split fractions must be finite in [0,1]') if not math.isclose(sum((self.burn, self.validators, self.treasury)), 1.0, abs_tol=1e-9): raise ValueError('Split fractions must sum to 1') @dataclass(frozen=True) class Policy: name: str worker: Split window: Split def presets() -> list[Policy]: burn = Split(1, 0, 0) validator = Split(0, 1, 0) treasury = Split(0, 0, 1) return [ Policy('burn_100', burn, burn), Policy('validators_100', validator, validator), Policy('treasury_100', treasury, treasury), Policy('validators_80_treasury_20', Split(0, .8, .2), Split(0, .8, .2)), Policy('validators_70_treasury_25_burn_5', Split(.05, .7, .25), Split(.05, .7, .25)), Policy('worker_treasury_window_validators', treasury, validator), ] def parse_split(value: str) -> Split: try: parts = [float(x.strip()) for x in value.split(',')] except ValueError as ex: raise ValueError('Expected three comma-separated percentages (burn,validators,treasury)') from ex if len(parts) != 3: raise ValueError('Expected exactly three percentages (burn,validators,treasury)') if not math.isclose(sum(parts), 100, rel_tol=0, abs_tol=1e-7): raise ValueError('Percentages must sum to 100') return Split(*(p / 100 for p in parts)) def _read_json(data: bytes, name: str) -> list[dict]: try: doc = json.loads(data) except (ValueError, UnicodeError) as ex: raise ValueError(f'{name}: invalid JSON: {ex}') from ex if not isinstance(doc, dict) or not isinstance(doc.get('scenarios'), list) or not doc['scenarios']: raise ValueError(f'{name}: expected JSON containing nonempty scenarios[]') return doc['scenarios'] def load_sources(path: Path) -> tuple[list[dict], list[str]]: """Never extract ZIP content; parse only recognized JSON entries by name.""" found = [] if path.is_dir(): for f in path.rglob('pop_cluster_*.json'): if not f.is_file() or f.stat().st_size > MAX_FILE_SIZE: raise ValueError(f'oversized/inaccessible input file {f}') found.append((str(f), f.read_bytes())) elif path.is_file() and path.suffix.lower() == '.zip': with zipfile.ZipFile(path) as archive: for item in archive.infolist(): if item.is_dir() or not Path(item.filename).name.startswith('pop_cluster_') or not item.filename.endswith('.json'): continue if item.file_size > MAX_FILE_SIZE: raise ValueError(f'ZIP member exceeds size limit: {item.filename}') found.append((item.filename, archive.read(item))) elif path.is_file() and path.name.startswith('pop_cluster_') and path.suffix.lower() == '.json': if path.stat().st_size > MAX_FILE_SIZE: raise ValueError('JSON file exceeds size limit') found.append((str(path), path.read_bytes())) else: raise ValueError('Provide ZIP, directory of pop_cluster_*.json, or a pop_cluster_*.json file') if not found or len(found) > MAX_INPUT_FILES: raise ValueError(f'Expected 1..{MAX_INPUT_FILES} pop_cluster_*.json files, got {len(found)}') scenarios = [] for name, blob in sorted(found): for source in _read_json(blob, name): scenarios.append(source) return scenarios, [name for name, _ in sorted(found)] def _number(v: object, desc: str) -> float: if type(v) not in (int, float) or not math.isfinite(v): raise ValueError(f'missing/invalid {desc}') return float(v) def key_for(parameters: dict) -> str: """Research group with target worker count and seed removed.""" stable = {k: v for k, v in parameters.items() if k not in ('target_workers', 'seed')} return json.dumps(stable, sort_keys=True, separators=(',', ':')) def project(record: dict, policy: Policy, validators: int) -> dict: if not (isinstance(validators, int) and validators > 0): raise ValueError('validators must be positive') if not isinstance(record, dict): raise ValueError('invalid scenario') p, s, ops = (record.get('parameters'), record.get('summary'), record.get('operators')) if not isinstance(p, dict) or not isinstance(s, dict) or not isinstance(ops, list) or not ops: raise ValueError('scenario lacks parameters, summary or operators') worker = _number(s.get('burned_workers'), 'burned_workers') window = _number(s.get('burned_windows'), 'burned_windows') old_burn = _number(s.get('burned_total'), 'burned_total') minted = _number(s.get('minted'), 'minted') origin = _number(p.get('initial_balance'), 'initial_balance') * len(ops) miner_wallets = sum(_number(op.get('wallet'), 'operator.wallet') for op in ops) if min(worker, window, minted, origin, miner_wallets) < -1e-8: raise ValueError('negative source accounting amount') if abs(old_burn - worker - window) > TOLERANCE: raise ValueError('source fee components do not match') if abs((minted + origin - old_burn) - miner_wallets) > TOLERANCE: raise ValueError('source accounting not conserved (operator wallets vs mint minus fees)') burn = worker * policy.worker.burn + window * policy.window.burn paid_validators = worker * policy.worker.validators + window * policy.window.validators paid_treasury = worker * policy.worker.treasury + window * policy.window.treasury total_fees = worker + window if not math.isclose(total_fees, burn + paid_validators + paid_treasury, rel_tol=1e-10, abs_tol=1e-6): raise AssertionError('fee destination split violated conservation') outstanding = minted + origin - burn cash_with_users_and_validators = miner_wallets + paid_validators if not math.isclose(outstanding, cash_with_users_and_validators + paid_treasury, rel_tol=1e-10, abs_tol=TOLERANCE): raise AssertionError('post-fee balances violate conservation') # A fee diverted to an external account remains spent by its originating miner. return { 'policy': policy.name, 'group': key_for(p), 'target_workers': p.get('target_workers'), 'seed': p.get('seed'), 'minted': minted, 'genesis': origin, 'gross_supply': minted + origin, 'fees_worker': worker, 'fees_window': window, 'fees_total': total_fees, 'burned': burn, 'burn_pct_minted': 100 * burn / minted if minted else 0.0, 'paid_validators': paid_validators, 'validator_each': paid_validators / validators, 'paid_treasury': paid_treasury, 'outstanding_supply': outstanding, 'miner_wallets': miner_wallets, 'non_treasury_balances': cash_with_users_and_validators, 'treasury_share_outstanding_pct': (100 * paid_treasury / outstanding if outstanding else 0.0), 'target_miner_net': _number(s.get('target_net'), 'target_net'), 'target_win_pct': _number(s.get('target_win_percent'), 'target_win_percent'), 'gini_miners_only': _number(s.get('gini_operator_wallets'), 'gini_operator_wallets'), 'validators': validators, 'operators': len(ops), 'other_workers': p.get('others_workers'), 'worker_fee_scale': p.get('worker_fee_scale'), 'newcomers': p.get('newcomers'), 'days': p.get('days'), } def summarize(records: list[dict]) -> list[dict]: by = defaultdict(list) for r in records: by[(r['group'], r['target_workers'], r['policy'])].append(r) metrics = ('minted', 'fees_worker', 'fees_window', 'fees_total', 'burned', 'burn_pct_minted', 'paid_validators', 'validator_each', 'paid_treasury', 'outstanding_supply', 'miner_wallets', 'non_treasury_balances', 'treasury_share_outstanding_pct', 'target_miner_net', 'target_win_pct', 'gini_miners_only') rows = [] for (_, _, _), v in sorted(by.items(), key=lambda kv: (kv[1][0]['other_workers'], kv[1][0]['worker_fee_scale'], kv[0][1], kv[0][2])): r = v[0] rows.append({k: r[k] for k in ('group', 'target_workers', 'policy', 'other_workers', 'worker_fee_scale', 'newcomers', 'days', 'validators', 'operators')} | {'seeds': len(v), 'seed_list': ','.join(str(x['seed']) for x in v)} | {k: statistics.mean(x[k] for x in v) for k in metrics}) return rows def _format_line(r: dict) -> str: return (f"{r['target_workers']:>2} {r['policy']:<38} " f"{r['fees_total']:>11,.1f} {r['burned']:>11,.1f} " f"{r['paid_validators']:>11,.1f} {r['paid_treasury']:>11,.1f} " f"{r['outstanding_supply']:>13,.1f} {r['target_miner_net']:>12,.1f}") def output(out: Path, rows: list[dict], runs: list[dict], inputs: list[str]) -> list[Path]: out.mkdir(parents=True, exist_ok=True) t, c, d, j, z = [out / f for f in ('fee_distribution_summary.txt', 'fee_distribution_comparison.csv', 'fee_distribution_per_run.csv', 'fee_distribution.json', 'potatium-fee-analysis.zip')] for dest, content in ((c, rows), (d, runs)): with dest.open('w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=list(content[0])) writer.writeheader() writer.writerows(content) j.write_text(json.dumps({'inputs': inputs, 'aggregates': rows, 'runs': runs}, indent=2, ensure_ascii=False) + '\n', encoding='utf-8') lines = ['POTATIUM | Gebuehrenverteilung | FORSCHUNG, KEINE PROTOKOLLREGEL', 'Gleiche Mining-Ergebnisse je Szenario; nur das Ziel der Gebuehren wird umgebucht.', 'Validatoren sind hier EXTERNE Empfaenger und nehmen selbst NICHT am Mining teil.', 'Fonds-Guthaben ist nicht verbrannt, aber im Modell vorerst nicht ausgabefreigegeben.', 'Guthaben der Miner / Mining-Strategien bleiben daher bei allen Policies identisch.', 'Alle Betragswerte in PTM. Float nur fuer Forschungsrechnung.', ''] groups = defaultdict(list) for r in rows: groups[r['group']].append(r) for ix, (_, group) in enumerate(groups.items(), 1): first = group[0] lines.extend([f'SZENARIO {ix}: {first["operators"]} Betreiber (inkl. Neuer), ' f'andere Worker={first["other_workers"]}, ' f'Tariffaktor={first["worker_fee_scale"]}, Neue={first["newcomers"]}, ' f'Tage={first["days"]}, je W {first["seeds"]} Seeds', 'W Gebuehrenziel Gebuehren Burn Validatoren Fonds Supply gesamt Netto Miner']) # Only representative worker counts to keep TXT manageable. workers = sorted({r['target_workers'] for r in group}) selected = {x for x in (0, 3, 7, 8, 10) if x in workers} if not selected: selected = set(workers) for r in group: if r['target_workers'] in selected: lines.append(_format_line(r)) lines.append('') lines += ['GRENZEN / KEIN BEWEIS FUER VERTEILUNGSGERECHTIGKEIT:', '* Alle Gebuehren sind bereits in den Simulator-Ergebnissen bezahlt.', '* Validator-/Fonds-Zufluesse werden nicht als Miningbudget wieder ausgegeben.', '* Validatoren koennen real auch Miner sein; Rueckfluesse muessen separat simuliert werden.', '* Administrations-/Treasury-Governance, Transfers, Sybil und reale CPU fehlen.', '* Der Fonds kann einen grossen Teil der PTM blockieren, obwohl nichts verbrannt wird.', '* Bei anderen Verwendungszwecken und geaenderten Anreizen muss PoP neu simuliert werden.', '* Keine verbindliche Coin-/Gebuehrenregel, kein On-chain-Float.'] t.write_text('\n'.join(lines) + '\n', encoding='utf-8') with zipfile.ZipFile(z, 'w', compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive: for f in (t, c, d, j): archive.write(f, arcname=f.name) return [t, c, d, j, z] def main() -> None: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument('--input', type=Path, required=True, help='prior ZIP, JSON file, or directory with pop_cluster_*.json') parser.add_argument('--out', type=Path, default=Path.home() / 'potatium-fee-analysis') parser.add_argument('--validators', type=int, default=4, help='separate validator recipients in this accounting test') parser.add_argument('--custom-worker', type=str, help='percentage triplet burn,validator,treasury, e.g. 0,50,50') parser.add_argument('--custom-window', type=str, help='percentage triplet; if omitted same as --custom-worker') args = parser.parse_args() try: if args.validators < 1 or args.validators > 100: raise ValueError('validators must be between 1 and 100') policies = presets() if args.custom_worker or args.custom_window: w = parse_split(args.custom_worker or args.custom_window) f = parse_split(args.custom_window or args.custom_worker) policies.append(Policy('custom', w, f)) sources, files = load_sources(args.input) runs = [project(source, policy, args.validators) for source in sources for policy in policies] rows = summarize(runs) paths = output(args.out, rows, runs, files) except (ValueError, OSError, zipfile.BadZipFile, KeyError) as ex: parser.error(str(ex)) print(f'Gelesen: {len(files)} JSON-Datei(en), {len(sources)} simulierte Szenarien, {len(policies)} Gebührenmodelle.') for f in paths: print(f' {f}') print('Sende mir vorzugsweise potatium-fee-analysis.zip zur Auswertung.') if __name__ == '__main__': main()