michael/potatium
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: update Potatium v0.6 architecture and PoP research

Browse Source
This commit is contained in:
Michael Böhmländer committed 2026-10-25 11:45:32 +00:00
1 parent 75df8a5c0d
commit 862ccf1d03
34 files changed
+3257 -18
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/analyze_pop_fee_flows.py
+301
View File
@@ -0,0 +1,301 @@
#!/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()
scripts/simulate_pop_cluster.py
+556
View File
@@ -0,0 +1,556 @@
#!/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()
scripts/simulate_pop_dynamic.py
+532
View File
@@ -0,0 +1,532 @@
#!/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()
scripts/simulate_pop_longterm.py
+305
View File
@@ -0,0 +1,305 @@
#!/usr/bin/env python3
"""Potatium PoP Langzeitsimulation. Forschungsmodell, KEINE Konsensregeln.
Ein Mineur = eine freigegebene Identitaet. Hier keine Sybil-Angriffe,
keine echten Hashzeiten oder Validator-Zensur. Float nur OFF-CHAIN.
"""
from __future__ import annotations
import argparse
import csv
import json
import math
import random
import statistics
from dataclasses import dataclass, asdict
from pathlib import Path
PROFILES = {
"original": (0.70, 0.70, 0.70),
"balanced": (0.55, 0.30, 0.20),
"cautious": (0.50, 0.30, 0.20),
}
# Unveraenderte Beispieltarife aus v0.2; nur Progression skaliert.
FEE_MODELS = {"original": 1.0, "mild": 0.1, "low": 0.01}
HISTORY_DAYS = 90
def progressive_fee(rewards_90d: float) -> float:
if rewards_90d < 0:
raise ValueError("90-Tage-Ertrag darf nicht negativ sein")
return (min(rewards_90d, 10.0) * 0.00005
+ min(max(rewards_90d - 10.0, 0.0), 90.0) * 0.0002
+ max(rewards_90d - 100.0, 0.0) * 0.0005)
def fee_for_window(window: int, rewards_90d: float, fee_model: str) -> float:
if fee_model not in FEE_MODELS:
raise ValueError("Unbekanntes Gebuehrenmodell")
if window == 1:
return 0.0
p = progressive_fee(rewards_90d) * FEE_MODELS[fee_model]
if window == 2:
return 0.001 + p
if window == 3:
return 0.003 + 2 * p
raise ValueError("Nur Fenster 1 bis 3 erlaubt")
def gini(values: list[float]) -> float:
values = sorted(values)
if not values or sum(values) <= 0:
return 0.0
n = len(values)
return (2.0 * sum((i + 1) * v for i, v in enumerate(values)) / (n * sum(values))) - (n + 1) / n
def policy_for(miner_id: int, strategy: str) -> str:
if strategy != "mixed":
return strategy
return ("free", "rational", "rational", "always")[miner_id % 4]
def should_buy(policy: str, window: int, balance: float,
last90: float, fee_model: str, p: tuple[float, float, float],
share_value: float) -> tuple[bool, str]:
fee = fee_for_window(window, last90, fee_model)
if balance + 1e-12 < fee:
return False, "budget"
if policy == "free":
return False, "strategy"
if policy == "always":
return True, "bought"
if policy == "rational":
if window == 3:
expected_gain = p[2] * share_value - fee
else:
# Schaetzung: Kann nach einem weiteren Fehlschlag ein drittes Fenster kaufen.
future_fee = fee_for_window(3, last90, fee_model)
future_value = 0.0
if balance - fee + 1e-12 >= future_fee:
future_value = max(0.0, p[2] * share_value - future_fee)
expected_gain = p[1] * share_value + (1.0 - p[1]) * future_value - fee
return expected_gain > 0.0, "bought" if expected_gain > 0.0 else "strategy"
raise ValueError("Unbekannte Strategie")
@dataclass
class Miner:
ident: int
policy: str
balance: float = 0.0
earned: float = 0.0
paid: float = 0.0
recent: float = 0.0
wins: int = 0
tickets: int = 0
paid_windows2: int = 0
paid_windows3: int = 0
skipped_budget: int = 0
skipped_strategy: int = 0
def simulate(miners: int, days: int, seed: int, profile: str,
strategy: str, fee_model: str, reward: float,
rounds_per_day: int, initial_balance: float = 0.0) -> dict:
if miners < 1 or days < 1 or rounds_per_day < 1 or reward <= 0 or initial_balance < 0:
raise ValueError("Ungueltige Parameter")
if profile not in PROFILES or fee_model not in FEE_MODELS:
raise ValueError("Unbekanntes Profil oder Gebuehrenmodell")
if strategy not in ("free", "always", "rational", "mixed"):
raise ValueError("Unbekannte Strategie")
p = PROFILES[profile]
rng = random.Random(seed)
persons = [Miner(i + 1, policy_for(i, strategy), balance=initial_balance)
for i in range(miners)]
history = [[0.0] * HISTORY_DAYS for _ in persons]
issued = burned = 0.0
no_ticket_rounds = total_tickets = 0
q_estimate = p[0]
daily = []
for day in range(days):
idx = day % HISTORY_DAYS
for i, person in enumerate(persons):
person.recent = max(0.0, person.recent - history[i][idx])
history[i][idx] = 0.0
daily_wins = daily_burn = daily_tickets = 0
for _ in range(rounds_per_day):
winners_candidates = []
# Strategie nutzt nur vergangene, nicht zukuenftige Ticket-Daten.
share_value = reward / (1.0 + (miners - 1) * q_estimate)
for i, person in enumerate(persons):
qualified = rng.random() < p[0]
if not qualified:
for window in (2, 3):
buy, reason = should_buy(person.policy, window, person.balance,
person.recent, fee_model, p, share_value)
if not buy:
if reason == "budget":
person.skipped_budget += 1
else:
person.skipped_strategy += 1
break
fee = fee_for_window(window, person.recent, fee_model)
person.balance -= fee
person.paid += fee
burned += fee
daily_burn += fee
if window == 2:
person.paid_windows2 += 1
else:
person.paid_windows3 += 1
if rng.random() < p[window - 1]:
qualified = True
break
if qualified:
winners_candidates.append(i)
person.tickets += 1
total_tickets += len(winners_candidates)
daily_tickets += len(winners_candidates)
q_estimate = 0.995 * q_estimate + 0.005 * (len(winners_candidates) / miners)
if winners_candidates:
chosen = persons[rng.choice(winners_candidates)]
chosen.wins += 1
chosen.earned += reward
chosen.balance += reward
chosen.recent += reward
history[chosen.ident - 1][idx] += reward
issued += reward
daily_wins += 1
else:
no_ticket_rounds += 1
daily.append({
"day": day + 1,
"issued_total": round(issued, 8),
"burned_total": round(burned, 8),
"supply": round(miners * initial_balance + issued - burned, 8),
"winners": daily_wins,
"tickets": daily_tickets,
"fees_burned": round(daily_burn, 8),
"gini_wallet": round(gini([m.balance for m in persons]), 6),
"q_estimate": round(q_estimate, 6),
})
balances = [max(0.0, m.balance) for m in persons]
total = sum(balances)
richest_count = max(1, math.ceil(miners * 0.10))
top_share = sum(sorted(balances, reverse=True)[:richest_count]) / total if total > 0 else 0
rounds = days * rounds_per_day
groups = {}
for group in ("free", "always", "rational"):
members = [m for m in persons if m.policy == group]
if members:
groups[group] = {
"miners": len(members),
"avg_wins": round(statistics.mean(m.wins for m in members), 3),
"avg_earned": round(statistics.mean(m.earned for m in members), 6),
"avg_fees_paid": round(statistics.mean(m.paid for m in members), 6),
"avg_balance": round(statistics.mean(m.balance for m in members), 6),
"avg_90d_rewards": round(statistics.mean(m.recent for m in members), 6),
"window2_per_miner": round(statistics.mean(m.paid_windows2 for m in members), 3),
"window3_per_miner": round(statistics.mean(m.paid_windows3 for m in members), 3),
}
summary = {
"model": "RESEARCH ONLY, not blockchain consensus",
"parameters": {
"miners": miners, "days": days, "seed": seed, "profile": profile,
"probabilities": p, "strategy": strategy, "fee_model": fee_model,
"reward": reward, "rounds_per_day": rounds_per_day,
"initial_balance": initial_balance,
"90d_history": HISTORY_DAYS,
},
"rounds_total": rounds,
"issued": round(issued, 8),
"burned": round(burned, 8),
"circulating": round(miners * initial_balance + issued - burned, 8),
"no_ticket_rounds": no_ticket_rounds,
"ticket_rate": round(total_tickets / (rounds * miners), 6),
"gini_wallet": round(gini(balances), 6),
"share_top10pct_wallet": round(top_share, 6),
"mean_wallet": round(statistics.mean(balances), 6),
"median_wallet": round(statistics.median(balances), 6),
"groups": groups,
"warnings": [
"Rechenarbeit nur als unabhaengige Bernoulli-Zufallsaufgabe modelliert; keine Hashzeiten.",
"Eine Identitaet = ein Teilnehmer, keine Sybil-/Mehrgeraeteangriffe oder Wallet-Verschiebung.",
"Strategie 'rational' verwendet eine einfache Nutzen-Schaetzung, keine optimale Spieltheorie.",
"Gebuehren sind Forschungswerte. Float niemals fuer Konsensberechnungen verwenden.",
"Keine Validator-Zensur, Kosten fuer Strom, Preisentwicklung oder Netzwerk-Ausfaelle.",
],
}
# Invariants fuer Simulation; tolerante Float-Pruefung (nie On-chain).
assert all(m.balance >= -1e-8 for m in persons)
assert math.isclose(sum(m.earned for m in persons), issued, abs_tol=1e-6)
assert math.isclose(sum(m.paid for m in persons), burned, abs_tol=1e-6)
assert math.isclose(sum(m.balance for m in persons), miners * initial_balance + issued - burned,
abs_tol=max(1e-6, (miners * initial_balance + issued) * 1e-9))
return {"summary": summary, "miners": [asdict(m) for m in persons], "daily": daily}
def save_results(out_dir: Path, result: dict) -> list[Path]:
out_dir.mkdir(parents=True, exist_ok=True)
s = result["summary"]
p = s["parameters"]
prefix = f"pop_{p['miners']}m_{p['days']}d_{p['profile']}_{p['strategy']}_{p['fee_model']}_{p['seed']}"
json_path = out_dir / (prefix + ".json")
txt_path = out_dir / (prefix + ".txt")
miners_csv = out_dir / (prefix + "_miners.csv")
daily_csv = out_dir / (prefix + "_daily.csv")
json_path.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
with miners_csv.open("w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=result["miners"][0].keys())
w.writeheader()
w.writerows(result["miners"])
with daily_csv.open("w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=result["daily"][0].keys())
w.writeheader()
w.writerows(result["daily"])
lines = ["POTATIUM · Langzeitsimulation | FORSCHUNG, KEINE PROTOKOLLREGEL",
"", f"Miner: {p['miners']} | Tage: {p['days']} | Runden/Tag: {p['rounds_per_day']}",
f"Erfolg pro Fenster: {p['probabilities']} | Strategie: {p['strategy']} | Gebuehren: {p['fee_model']}",
f"Seed: {p['seed']} | Brutto-Belohnung: {p['reward']} PTM", "",
f"Runden: {s['rounds_total']:,} | ohne Tickets: {s['no_ticket_rounds']:,}",
f"Ticketquote: {100*s['ticket_rate']:.2f}% | Gini Guthaben: {s['gini_wallet']:.3f}",
f"Erzeugt: {s['issued']:.4f} PTM | Verbrannt: {s['burned']:.4f} PTM | Umlauf: {s['circulating']:.4f} PTM",
f"Top 10 % Guthabenanteil: {100*s['share_top10pct_wallet']:.2f}%", "",
"Strategie Miner Ø Gewinne Ø Gebühren Ø Wallet Ø 90d-Gewinn Ø F2 Ø F3"]
for name, g in s["groups"].items():
lines.append(f"{name:<14} {g['miners']:>5} {g['avg_wins']:>10.2f} {g['avg_fees_paid']:>12.4f} "
f"{g['avg_balance']:>11.4f} {g['avg_90d_rewards']:>13.2f} "
f"{g['window2_per_miner']:>5.1f} {g['window3_per_miner']:>5.1f}")
lines += ["", "MODELLGRENZEN:"] + ["- " + w for w in s["warnings"]]
txt_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
print("\n".join(lines))
return [txt_path, json_path, miners_csv, daily_csv]
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--miners", type=int, default=20)
ap.add_argument("--days", type=int, default=180)
ap.add_argument("--rounds-per-day", type=int, default=720,
help="720 bei 2 Minuten je Runde; kleiner fuer schnelle Tests")
ap.add_argument("--seed", type=int, default=20261025)
ap.add_argument("--profile", choices=PROFILES, default="balanced")
ap.add_argument("--strategy", choices=["free", "always", "rational", "mixed"], default="mixed")
ap.add_argument("--fee-model", choices=FEE_MODELS, default="mild")
ap.add_argument("--reward", type=float, default=1.0)
ap.add_argument("--initial-balance", type=float, default=0.0)
ap.add_argument("--out", type=Path, default=Path.home() / "potatium-simulation")
args = ap.parse_args()
result = simulate(args.miners, args.days, args.seed, args.profile, args.strategy,
args.fee_model, args.reward, args.rounds_per_day, args.initial_balance)
paths = save_results(args.out, result)
print("\nErstellt:")
for path in paths:
print(path)
if __name__ == "__main__":
main()