306 lines
13 KiB
Python
306 lines
13 KiB
Python
#!/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()
|