x402_measure.py

raw

#!/usr/bin/env python3
"""
Independently measure gasless (EIP-3009) USDC settlement on Base — the mechanism x402
uses to move money — by counting AuthorizationUsed events straight from chain.

Samples several windows across ~24h rather than extrapolating from one, because activity
is bursty. Reports what it measured and, importantly, what it does NOT establish.
"""
import json, time, urllib.request, statistics

RPCS = ["https://mainnet.base.org", "https://base-rpc.publicnode.com", "https://base.drpc.org"]
USDC = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
TOPIC = "0x98de503528ee59b575ef0c0a2576a82497bfc029a5685b209e9ec333479b10a5"  # AuthorizationUsed(address,bytes32)
SPAN = 500          # blocks per sample; public RPCs reject much more
BLOCK_SECONDS = 2.0  # Base
OUT = "/var/www/aaw/status/x402.json"
UA = {"User-Agent": "Mozilla/5.0 (compatible; agentatwork/1.0)"}


def rpc(method, params):
    body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}).encode()
    last = None
    for url in RPCS:
        try:
            req = urllib.request.Request(url, body, {"content-type": "application/json", **UA})
            with urllib.request.urlopen(req, timeout=30) as r:
                d = json.load(r)
            if "result" in d:
                return d["result"]
            last = d.get("error")
        except Exception as e:
            last = str(e)
    raise RuntimeError(f"rpc failed: {last}")


def main():
    head = int(rpc("eth_blockNumber", []), 16)
    day = int(86400 / BLOCK_SECONDS)                       # ~43200 blocks
    offsets = [0, day // 6, day // 3, day // 2, 2 * day // 3, 5 * day // 6, day]
    samples = []

    for off in offsets:
        hi = head - off
        lo = hi - SPAN
        try:
            logs = rpc("eth_getLogs", [{"address": USDC, "topics": [TOPIC],
                                        "fromBlock": hex(lo), "toBlock": hex(hi)}])
            per_min = len(logs) / (SPAN * BLOCK_SECONDS / 60)
            samples.append({"from_block": lo, "to_block": hi, "events": len(logs),
                            "hours_ago": round(off * BLOCK_SECONDS / 3600, 1),
                            "per_minute": round(per_min, 1)})
            print(f"  blocks {lo}-{hi} ({off*BLOCK_SECONDS/3600:5.1f}h ago): "
                  f"{len(logs):>6} events  = {per_min:7.1f}/min")
        except Exception as e:
            print(f"  offset {off}: {e}")
        time.sleep(1.5)

    if not samples:
        raise SystemExit("no samples")

    rates = [s["per_minute"] for s in samples]
    median = statistics.median(rates)
    data = {
        "measured_utc": time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime()),
        "chain": "Base", "token": "USDC", "event": "AuthorizationUsed (EIP-3009)",
        "head_block": head,
        "samples": samples,
        "median_per_minute": round(median, 1),
        "implied_per_day": int(median * 60 * 24),
        "implied_per_30d": int(median * 60 * 24 * 30),
        "caveats": [
            "EIP-3009 gasless authorizations are the mechanism x402 settles with, but not "
            "exclusively x402 — other gasless USDC flows use the same event.",
            "Base only. x402 also settles on Solana, which this does not count.",
            "Sampled windows of 500 blocks, not a full census. Activity is bursty.",
        ],
    }
    json.dump(data, open(OUT, "w"), indent=1)
    print(f"\nmedian {median:.1f}/min -> ~{data['implied_per_day']:,}/day "
          f"-> ~{data['implied_per_30d']:,}/30d")


if __name__ == "__main__":
    main()