monitoring: final stats dashboard and ladder reserve matching

Print a boxed FINAL STATISTICS block after e2e/ladder (coins, flow, tendencies, perf, load) and match force-select reserve_pub to the current withdrawal so cumulative ladder wallets do not reuse stale pubs.
2026-07-17 08:28:00 +02:00 · 2026-07-17 08:28:00 +02:00 · b65018cd0b
commit b65018cd0b
parent 48f77139c5
4 changed files with 609 additions and 105 deletions
--- a/scripts/taler-monitoring/README.md
+++ b/scripts/taler-monitoring/README.md
@ -126,6 +126,20 @@ Also: coin counts (in circulation / spent), status histogram, Δ vs previous sna
 History TSV: `$METRICS_DIR/coins-history.tsv`. Map loaded from
 `${EXCHANGE_PUBLIC}/config` → `currency_specification.alt_unit_names`.

+### Final statistics dashboard (e2e · ladder)
+
+At the end of **e2e** and **ladder**, a boxed **FINAL STATISTICS** block prints:
+
+| Section | Content |
+|---------|---------|
+| **Coins** | in circulation / spent counts + amounts + denoms (alt names) |
+| **Money flow** | withdrawn events/amount, spent (paid), net (wd − spent) |
+| **Tendency** | coin/amount trend over snaps (↑↓→), slope, last history rows |
+| **Performance** | min / p50 / avg / max per phase; first-half vs second-half latency |
+| **Load** | host loadavg + mem before/after; per-container RSS/CPU; tendencies |
+
+Successful withdraw/pay steps append to `flow-withdrawn.txt` / `flow-spent.txt` (ladder also uses its TSVs).
+
 ## Load / memory (inside · e2e · ladder)

 Host **loadavg**, **RAM used/avail**, and per-container **RSS / CPU / block I/O**
--- a/scripts/taler-monitoring/check_e2e.sh
+++ b/scripts/taler-monitoring/check_e2e.sh
@ -152,18 +152,14 @@ e2e_finish() {
  fi
  E2E_REPORTED=1
  print_balances
-  # Final coin inventory + host/container load
+  # Final coin inventory + host/container load + statistics dashboard
  if [ -n "${METRICS_DIR:-}" ]; then
    section "metrics · e2e coins final"
    metrics_report_coins "e2e-end" || true
-  fi
-  if [ -n "${METRICS_DIR:-}" ] && [ "${METRICS_LOAD:-1}" != "0" ]; then
+    if [ "${METRICS_LOAD:-1}" != "0" ]; then
      metrics_report_load "${METRICS_DIR}/load-after.json" "e2e-end" || true
-    if [ -f "${METRICS_DIR}/load-before.json" ] && [ -f "${METRICS_DIR}/load-after.json" ]; then
-      section "metrics · e2e load delta"
-      metrics_print_load_delta "${METRICS_DIR}/load-before.json" "${METRICS_DIR}/load-after.json" || true
    fi
-    metrics_print_overall "e2e overall" || true
+    metrics_print_overall "e2e final" || true
  fi
  summary || true
  return "$code"
@ -798,6 +794,7 @@ else:
  done
  metrics_report_coins "after-ATM-${tag}" || true
  if [ "$ok_bal" = "1" ]; then
+    metrics_record_flow withdrawn "$WITHDRAW_AMT" || true
    return 0
  fi
  # Bank side often already confirmed — treat as timing lag, not hard fail
@ -897,6 +894,7 @@ sys.exit(0 if d.get("paid") is True or str(d.get("order_status","")).lower()=="p
 ' "$SCRATCH/ord-paid-$tag.json" 2>/dev/null; then
      ok "payment settled $PAY_AMT ($PAY_SUM · order $OID)"
      metrics_report_coins "after-pay-${tag}" || true
+      metrics_record_flow spent "$PAY_AMT" || true
      return 0
    fi
  done
@ -1017,6 +1015,7 @@ sys.exit(0 if d.get("paid") is True or str(d.get("order_status","")).lower()=="p
 ' "$SCRATCH/ord-paid-$tag.json" 2>/dev/null; then
        ok "shop $pname" "payment settled ($pamt · order $OID)"
        metrics_report_coins "after-shop-${tag}" || true
+        metrics_record_flow spent "$pamt" || true
        return 0
      fi
    fi
@ -1024,6 +1023,7 @@ sys.exit(0 if d.get("paid") is True or str(d.get("order_status","")).lower()=="p
      || grep -qiE 'done|paid|success|Payment' "$SCRATCH/pay-$tag.out" 2>/dev/null; then
      ok "shop $pname" "payment settled via wallet tx ($pamt · order $OID)"
      metrics_report_coins "after-shop-${tag}" || true
+      metrics_record_flow spent "$pamt" || true
      return 0
    fi
  done
--- a/scripts/taler-monitoring/check_goa_ladder.sh
+++ b/scripts/taler-monitoring/check_goa_ladder.sh
@ -483,58 +483,138 @@ for AMT in "$@"; do
      -H "Authorization: Bearer ${TOK}" -H 'Content-Type: application/json' -d '{}' \
      "${BANK}/accounts/${EXP_USER}/withdrawals/${WID}/confirm"
  }
-  extract_rpub() {
-    # Prefer *last* match (current withdraw), not first (stale from older accepts/tx).
-    python3 -c '
-import re, sys, json
-paths = sys.argv[1:]
-found = []
-blob = ""
-for p in paths:
-    try:
-        blob += open(p, errors="replace").read() + "\n"
-    except Exception:
-        pass
-for pat in (
-    r"\"reserve_pub\"\s*:\s*\"([A-Z0-9]+)\"",
-    r"\"reservePub\"\s*:\s*\"([A-Z0-9]+)\"",
-    r"reserve_pub[\"\s:=]+([A-Z0-9]{40,})",
-    r"reservePub[\"\s:=]+([A-Z0-9]{40,})",
-    r"reserve[_ ]?pub[\"=: ]+([A-Z0-9]{40,})",
-):
-    found.extend(m.group(1) for m in re.finditer(pat, blob, re.I))
-def walk(o):
+  # Collect reserve_pub candidates for *this* withdrawal (WID + amount).
+  # Cumulative wallets re-print old reserves in accept/tx dumps — never trust a single "last" blindly.
+  # Prints unique pubs one per line, preferred order first.
+  extract_rpubs_for_wid() {
+    python3 - "$WID" "$AMT" "$SCRATCH/accept-$tag.out" "$SCRATCH/tx-$tag.json" "$SCRATCH/used-rpubs.txt" <<'PY'
+import json, re, sys
+from pathlib import Path
+
+wid = sys.argv[1]
+amt = sys.argv[2]
+paths = sys.argv[3:5]
+used_path = sys.argv[5]
+used = set()
+if Path(used_path).is_file():
+    used = {ln.strip() for ln in open(used_path) if ln.strip()}
+
+def walk_collect(o, bag, ctx=None):
+    ctx = dict(ctx or {})
    if isinstance(o, dict):
        for k, v in o.items():
-            if k.lower() in ("reserve_pub", "reservepub") and isinstance(v, str) and len(v) >= 40:
-                found.append(v)
-            walk(v)
+            kl = str(k).lower()
+            if kl in ("withdrawal_id", "withdraw_id", "wopid", "id") and isinstance(v, str):
+                ctx["id"] = v
+            if kl in ("amount", "rawamount", "instructedamount") and isinstance(v, str):
+                ctx["amount"] = v
+            if kl in ("taler_withdraw_uri", "talerwithdrawuri", "uri") and isinstance(v, str):
+                ctx["uri"] = v
+            if kl in ("reserve_pub", "reservepub") and isinstance(v, str) and len(v) >= 40:
+                bag.append((v, dict(ctx)))
+            walk_collect(v, bag, ctx)
    elif isinstance(o, list):
        for i in o:
-            walk(i)
-for line in blob.splitlines():
-    line = line.strip()
-    if not line.startswith("{"):
-        continue
+            walk_collect(i, bag, ctx)
+
+raw_pubs = []  # ordered as found
+scored = []    # (score, pub) higher better
+blob_all = ""
+for p in paths:
    try:
-        walk(json.loads(line))
+        blob_all += open(p, errors="replace").read() + "\n"
    except Exception:
        pass
-if found:
-    print(found[-1])
-' "$@" 2>/dev/null || true
+
+# 1) full JSON objects in files
+for p in paths:
+    try:
+        t = open(p, errors="replace").read()
+    except Exception:
+        continue
+    # whole-file JSON
+    for m in re.finditer(r"\{", t):
+        try:
+            o = json.loads(t[m.start():])
+        except Exception:
+            continue
+        bag = []
+        walk_collect(o, bag)
+        for pub, ctx in bag:
+            raw_pubs.append(pub)
+            score = 0
+            cid = str(ctx.get("id") or "")
+            camt = str(ctx.get("amount") or "")
+            curi = str(ctx.get("uri") or "")
+            if wid and wid in cid:
+                score += 100
+            if wid and wid in curi:
+                score += 80
+            if amt and (camt == amt or camt.endswith(amt.split(":", 1)[-1])):
+                score += 40
+            if pub in used:
+                score -= 200
+            scored.append((score, pub))
+
+# 2) regex fallback on accept file only (more current)
+try:
+    acc = open(paths[0], errors="replace").read()
+except Exception:
+    acc = ""
+for pat in (
+    r"\"reserve_pub\"\s*:\s*\"([A-Z0-9]{40,})\"",
+    r"\"reservePub\"\s*:\s*\"([A-Z0-9]{40,})\"",
+    r"reserve_pub[\"\s:=]+([A-Z0-9]{40,})",
+):
+    for m in re.finditer(pat, acc, re.I):
+        pub = m.group(1)
+        raw_pubs.append(pub)
+        score = 10
+        # proximity to WID in accept output
+        window = acc[max(0, m.start() - 400) : m.end() + 400]
+        if wid and wid in window:
+            score += 100
+        if amt and amt in window:
+            score += 30
+        if pub in used:
+            score -= 200
+        scored.append((score, pub))
+
+# prefer high score, then later occurrence
+order = []
+seen = set()
+for score, pub in sorted(scored, key=lambda x: (-x[0],), reverse=False):
+    # sort by score desc: use reverse sorted
+    pass
+for score, pub in sorted(scored, key=lambda x: x[0], reverse=True):
+    if pub in seen or pub in used:
+        continue
+    seen.add(pub)
+    order.append(pub)
+# append unused raw in reverse (newest-ish)
+for pub in reversed(raw_pubs):
+    if pub in seen or pub in used:
+        continue
+    seen.add(pub)
+    order.append(pub)
+for pub in order:
+    print(pub)
+PY
+  }
+
+  mark_rpub_used() {
+    local p="$1"
+    [ -n "$p" ] || return 0
+    mkdir -p "$SCRATCH" 2>/dev/null || true
+    grep -qxF "$p" "$SCRATCH/used-rpubs.txt" 2>/dev/null || echo "$p" >>"$SCRATCH/used-rpubs.txt"
  }

  force_select_if_needed() {
    local st_now="$1"
    [ "$st_now" = "pending" ] || [ -z "$st_now" ] || return 0
-    local rpub epayto code_fs
-    # current accept only first — avoid reusing reserve from previous rungs via tx dump
-    rpub=$(extract_rpub "$SCRATCH/accept-$tag.out")
-    if [ -z "$rpub" ]; then
+    local rpub epayto code_fs any=0
+    # refresh tx dump each try (wallet may attach reserve late)
    wcli transactions >"$SCRATCH/tx-$tag.json" 2>&1 || true
-      rpub=$(extract_rpub "$SCRATCH/tx-$tag.json")
-    fi
    epayto=$(curl -sS -m 10 "${EX%/}/keys" 2>/dev/null | python3 -c '
 import json,sys
 d=json.load(sys.stdin)
@ -546,31 +626,46 @@ for a in acc:
 else:
  if acc: print(acc[0].get("payto_uri") or "")
 ' 2>/dev/null || true)
-    if [ -n "$rpub" ] && [ -n "$epayto" ]; then
+    if [ -z "$epayto" ]; then
+      warn bank "force-select skipped" "problem: exchange payto empty from /keys"
+      return 0
+    fi
+    # Try candidates until bank leaves pending (200/204) or we exhaust
+    while IFS= read -r rpub; do
+      [ -n "$rpub" ] || continue
+      any=1
      code_fs=$(curl -sS -m 12 -o "$SCRATCH/force-sel-$tag.json" -w '%{http_code}' -X POST \
        -H 'Content-Type: application/json' \
        -d "{\"reserve_pub\":\"${rpub}\",\"selected_exchange\":\"${epayto}\"}" \
        "${BANK}/taler-integration/withdrawal-operation/${WID}" 2>/dev/null || echo "000")
-      # 409: conflict (wrong/stale reserve, or already bound) — log body once, keep polling
+      if [ "$code_fs" = "200" ] || [ "$code_fs" = "204" ]; then
+        info "force-select" "HTTP ${code_fs} rpub=${rpub:0:12}… (ok for WID ${WID:0:8})"
+        mark_rpub_used "$rpub"
+        return 0
+      fi
      if [ "$code_fs" = "409" ]; then
-        if [ "${FORCE_SEL_409_LOGGED:-0}" != "1" ]; then
-          info "force-select" "HTTP 409 rpub=${rpub:0:12}… body=$(tr '\n' ' ' <"$SCRATCH/force-sel-$tag.json" | head -c 160)"
-          FORCE_SEL_409_LOGGED=1
+        # 5114 = this reserve already bound to another op — not "out of money"
+        info "force-select" "HTTP 409 rpub=${rpub:0:12}… (stale/used reserve — not balance; trying next)"
+        mark_rpub_used "$rpub"
+        continue
      fi
-      else
-        info "force-select" "HTTP ${code_fs} rpub=${rpub:0:12}…"
-        FORCE_SEL_409_LOGGED=0
-      fi
-    else
-      _rpub_empty=no
-      [ -z "$rpub" ] && _rpub_empty=yes
-      _epayto_empty=no
-      [ -z "$epayto" ] && _epayto_empty=yes
+      info "force-select" "HTTP ${code_fs} rpub=${rpub:0:12}… body=$(tr '\n' ' ' <"$SCRATCH/force-sel-$tag.json" 2>/dev/null | head -c 120)"
+      # other errors: still try next candidate
+    done < <(extract_rpubs_for_wid)
+    if [ "$any" != "1" ]; then
      warn bank "force-select skipped" \
-        "problem: cannot move bank pending->selected (reserve_pub empty=${_rpub_empty}, exchange payto empty=${_epayto_empty})"
+        "problem: no reserve_pub for this withdraw (WID=${WID:0:8}…); wallet may not have selected yet"
    fi
  }

+  # Short bounded select assist (≤8s) — not a hang path; helps attach fresh reserve_pub
+  if command -v perl >/dev/null 2>&1 && [ -f "$CLI_JS" ]; then
+    perl -e 'alarm shift; exec @ARGV' 8 \
+      node "$CLI_JS" --wallet-db="$WDB" --no-throttle run-until-done \
+      >"$SCRATCH/select-$tag.out" 2>&1 || true
+  fi
+  wcli transactions >"$SCRATCH/tx-$tag.json" 2>&1 || true
+
  t0=$(now_ms)
  conf_ok=0
  st=""
@ -598,9 +693,9 @@ else:
        break
        ;;
    esac
-    # force-select while pending (no run-until-done); avoid spam on repeated 409
+    # force-select while pending — try alternate rpubs on 5114 (not out-of-money)
    if [ "$st" = "pending" ] || [ -z "$st" ]; then
-      if [ "$i" -eq 1 ] || [ "$i" -eq 2 ] || [ $((i % 5)) -eq 0 ]; then
+      if [ "$i" -eq 1 ] || [ "$i" -eq 2 ] || [ $((i % 3)) -eq 0 ]; then
        force_select_if_needed "$st"
      fi
    fi
@ -610,10 +705,10 @@ else:
  st=$(printf '%s' "${st:-}" | tr -d '\r\n' | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')
  if [ "$conf_ok" != "1" ]; then
    note="${note:-confirm timeout last=${st:-empty}}"
-    # Soft: not confirmed after polls — WARN and continue (pending/select lag or force-select issues)
+    # Soft: not confirmed — usually stale reserve_pub (5114), NOT empty pool balance
    status="SKIP_CONFIRM"
    warn bank "confirm $AMT skipped" \
-      "problem: bank status='${st:-empty}' after ${LADDER_CONFIRM_POLLS} polls (want selected/confirmed); ladder continues. detail: $note"
+      "problem: bank status='${st:-empty}' after ${LADDER_CONFIRM_POLLS} polls (want selected/confirmed). Usually reserve_pub mismatch (5114), not out-of-money — mint/accept already OK. detail: $note"
    ms_total=$(elapsed_ms "$t_rung")
    echo -e "${rung}\t${range_note}\t${AMT}\t${status}\t${ms_mint}\t${ms_accept}\t${ms_confirm}\t${ms_settle}\t${ms_total}\t${WID}\t${note}" >>"$TSV"
    continue
@ -664,6 +759,7 @@ sys.exit(0 if Decimal(sys.argv[1]) > Decimal(sys.argv[2]) else 1)
    OK_N=$((OK_N + 1))
    echo -e "${rung}\t${range_note}\t${AMT}\t${status}\t${ms_mint}\t${ms_accept}\t${ms_confirm}\t${ms_settle}\t${ms_total}\t${WID}\t${note}" >>"$TSV"
    metrics_report_coins "ladder-r${rung}-after-${tag}" || true
+    metrics_record_flow withdrawn "$AMT" || true
  elif echo "$xfer" | grep -qi True; then
    status="OK_BANK"
    note="bank transfer_done avail=${after} $xfer (no run-until-done)"
@ -671,6 +767,7 @@ sys.exit(0 if Decimal(sys.argv[1]) > Decimal(sys.argv[2]) else 1)
    OK_N=$((OK_N + 1))
    echo -e "${rung}\t${range_note}\t${AMT}\t${status}\t${ms_mint}\t${ms_accept}\t${ms_confirm}\t${ms_settle}\t${ms_total}\t${WID}\t${note}" >>"$TSV"
    metrics_report_coins "ladder-r${rung}-after-${tag}" || true
+    metrics_record_flow withdrawn "$AMT" || true
  else
    note="no coins / no transfer_done avail=${after} $xfer"
    err wallet "settle $AMT" "$note"
@ -897,6 +994,7 @@ sys.exit(0 if d.get("paid") is True or str(d.get("order_status","")).lower()=="p
        PAY_OK_N=$((PAY_OK_N + 1))
        echo -e "${prung}\t${prange}\t${PAMT}\t${pstatus}\t${ms_order}\t${ms_handle}\t${ms_psettle}\t${ms_total}\t${OID}\t${pnote}" >>"$PAY_TSV"
        metrics_report_coins "after-pay-${ptag}" || true
+        metrics_record_flow spent "$PAMT" || true
      else
        pnote="not settled order=$OID avail=${CUR}:${after}"
        if [ "$IS_PMAX" = "1" ]; then
@ -1048,7 +1146,10 @@ for k, v in (rep.get("timing") or {}).items():
 json.dump(out, open(sys.argv[2], "w"), indent=2)
 PY
 fi
-metrics_print_overall "ladder overall" || true
+export METRICS_WITHDRAW_TSV="$TSV"
+export METRICS_PAY_TSV="$PAY_TSV"
+metrics_report_coins "ladder-end" || true
+metrics_print_overall "ladder final" || true

 # Keep scratch if LADDER_REPORT_DIR set; else copy key files to /tmp
 if [ -z "${LADDER_REPORT_DIR:-}" ]; then
--- a/scripts/taler-monitoring/metrics.sh
+++ b/scripts/taler-monitoring/metrics.sh
@ -976,44 +976,433 @@ for role in ("bank","exchange","merchant"):
 PY
 }

+# Record a successful withdraw/pay amount for end statistics (one amount per line).
+# $1=withdrawn|spent  $2=amount (CUR:n)
+metrics_record_flow() {
+  local kind="$1" amt="$2"
+  [ -n "$amt" ] || return 0
+  mkdir -p "${METRICS_DIR}" 2>/dev/null || true
+  case "$kind" in
+    withdrawn|withdraw) printf '%s\n' "$amt" >>"${METRICS_DIR}/flow-withdrawn.txt" ;;
+    spent|pay)          printf '%s\n' "$amt" >>"${METRICS_DIR}/flow-spent.txt" ;;
+    *) return 1 ;;
+  esac
+}
+
 # Overall end-of-run statistics block
-# Args via env / files:
-#   METRICS_DIR, optional: WITHDRAW_REPORT, PAY_REPORT, phase timings JSON files
+# Files under METRICS_DIR:
+#   coins-final.json, coins-history.tsv, perf-summary.json,
+#   load-before/after.json, flow-withdrawn.txt, flow-spent.txt,
+#   optional ladder TSV via METRICS_WITHDRAW_TSV / METRICS_PAY_TSV
 metrics_print_overall() {
  local title="${1:-overall statistics}"
-  section "metrics · $title"
-  if [ -f "${METRICS_DIR}/load-before.json" ]; then
-    info "load BEFORE" ""
-    metrics_print_load "${METRICS_DIR}/load-before.json" "before"
-  fi
-  if [ -f "${METRICS_DIR}/load-after.json" ]; then
-    info "load AFTER" ""
-    metrics_print_load "${METRICS_DIR}/load-after.json" "after"
-    metrics_print_load_delta "${METRICS_DIR}/load-before.json" "${METRICS_DIR}/load-after.json"
-  fi
-  if [ -f "${METRICS_DIR}/coins-final.json" ]; then
-    info "coins final" "$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1])).get("summary","?"))' "${METRICS_DIR}/coins-final.json" 2>/dev/null || echo n/a)"
-  fi
-  if [ -f "${METRICS_DIR}/coins-history.tsv" ]; then
-    echo "  --- coins after each withdraw ---"
-    # header + rows
-    if [ -s "${METRICS_DIR}/coins-history.tsv" ]; then
-      column -t -s $'\t' "${METRICS_DIR}/coins-history.tsv" 2>/dev/null \
-        || cat "${METRICS_DIR}/coins-history.tsv"
-    fi
-  fi
-  if [ -f "${METRICS_DIR}/perf-summary.json" ]; then
-    info "performance" ""
-    python3 - "${METRICS_DIR}/perf-summary.json" <<'PY'
-import json,sys
-d=json.load(open(sys.argv[1]))
-for k,v in d.items():
-  if isinstance(v, dict) and "n" in v:
-    print(f"  {k:14} n={v.get('n')}  min={v.get('min_ms')}ms  p50={v.get('p50_ms')}ms  avg={v.get('avg_ms')}ms  max={v.get('max_ms')}ms")
-  else:
-    print(f"  {k}: {v}")
-PY
-  fi
-  # free-form extras from caller
+  section "statistics · $title"
+  metrics_print_final_stats || true
  [ -n "${METRICS_EXTRA_LINES:-}" ] && printf '%s\n' "$METRICS_EXTRA_LINES"
 }
+
+# Rich final dashboard: coins, withdrawn/spent, performance, tendencies
+metrics_print_final_stats() {
+  python3 - \
+    "${METRICS_DIR:-/tmp}" \
+    "${ALT_UNITS_FILE:-}" \
+    "${METRICS_WITHDRAW_TSV:-}" \
+    "${METRICS_PAY_TSV:-}" \
+    "${CUR:-GOA}" \
+    <<'PY'
+import csv, json, os, sys
+from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+from pathlib import Path
+
+mdir = Path(sys.argv[1])
+alt_path = sys.argv[2] or ""
+wd_tsv = sys.argv[3] or ""
+pay_tsv = sys.argv[4] or ""
+cur = sys.argv[5] or "GOA"
+
+alt = {}
+if alt_path and Path(alt_path).is_file():
+    try:
+        alt = json.load(open(alt_path))
+    except Exception:
+        alt = {}
+if not alt:
+    alt = {"0": cur}
+
+def D(x, default=Decimal(0)):
+    try:
+        return Decimal(str(x))
+    except Exception:
+        return default
+
+def parse_amt(s):
+    s = str(s or "").strip()
+    if not s:
+        return cur, Decimal(0)
+    if ":" in s:
+        c, v = s.split(":", 1)
+        try:
+            return c, Decimal(v)
+        except InvalidOperation:
+            return c, Decimal(0)
+    try:
+        return cur, Decimal(s)
+    except InvalidOperation:
+        return cur, Decimal(0)
+
+def fmt_num(v: Decimal) -> str:
+    if v == v.to_integral():
+        return format(int(v), "d")
+    s = format(v.quantize(Decimal("0.00000001"), rounding=ROUND_HALF_UP), "f")
+    return s.rstrip("0").rstrip(".")
+
+def fmt_amt(c, v: Decimal) -> str:
+    return "%s:%s" % (c, fmt_num(v))
+
+def fmt_alt(c, v: Decimal) -> str:
+    base_s = fmt_amt(c, v)
+    base_name = alt.get("0") or c
+    if v == 0:
+        return "0 %s (%s)" % (base_name, base_s)
+    scales = []
+    for k, name in alt.items():
+        try:
+            scales.append((int(k), str(name)))
+        except Exception:
+            pass
+    scales.sort(key=lambda x: -x[0])
+    absval = abs(v)
+    for sc, name in scales:
+        unit = Decimal(10) ** sc
+        coeff = absval / unit
+        if coeff >= 1:
+            if sc == 0:
+                return "%s %s" % (fmt_num(v), name)
+            return "%s %s (%s)" % (fmt_num(coeff if v >= 0 else -coeff), name, base_s)
+    return "%s %s (%s)" % (fmt_num(v), base_name, base_s)
+
+def sum_amount_file(path: Path):
+    total = Decimal(0)
+    ccy = cur
+    n = 0
+    if not path.is_file():
+        return ccy, total, n
+    for line in path.read_text().splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        c, v = parse_amt(line)
+        ccy = c
+        total += v
+        n += 1
+    return ccy, total, n
+
+def sum_tsv_amounts(path, amount_col="amount", status_ok=None):
+    total = Decimal(0)
+    ccy = cur
+    n = 0
+    if not path or not Path(path).is_file():
+        return ccy, total, n
+    with open(path, newline="") as f:
+        for row in csv.DictReader(f, delimiter="\t"):
+            st = (row.get("status") or "").upper()
+            if status_ok is not None and st not in status_ok:
+                continue
+            c, v = parse_amt(row.get(amount_col) or "")
+            if v <= 0:
+                continue
+            ccy = c
+            total += v
+            n += 1
+    return ccy, total, n
+
+def trend_label(delta, eps=Decimal("0.01")):
+    if delta > eps:
+        return "↑ rising"
+    if delta < -eps:
+        return "↓ falling"
+    return "→ stable"
+
+def half_trend(values):
+    """Compare avg of second half vs first half of a list of numbers."""
+    xs = [float(x) for x in values if x is not None]
+    if len(xs) < 4:
+        if len(xs) < 2:
+            return "n/a (few samples)", None, None
+        a, b = xs[0], xs[-1]
+        return trend_label(Decimal(str(b - a))), a, b
+    mid = len(xs) // 2
+    first = sum(xs[:mid]) / mid
+    second = sum(xs[mid:]) / (len(xs) - mid)
+    d = second - first
+    lab = "↑ slowing (worse)" if d > 0.05 * max(abs(first), 1) else (
+        "↓ faster (better)" if d < -0.05 * max(abs(first), 1) else "→ steady"
+    )
+    return lab, first, second
+
+print("")
+print("╔══════════════════════════════════════════════════════════════╗")
+print("║  FINAL STATISTICS                                            ║")
+print("╚══════════════════════════════════════════════════════════════╝")
+
+# --- coins snapshot ---
+coins = {}
+cf = mdir / "coins-final.json"
+if cf.is_file():
+    try:
+        coins = json.load(open(cf))
+    except Exception:
+        coins = {}
+
+print("")
+print("── Coins (wallet) ─────────────────────────────────────────────")
+if coins.get("ok"):
+    circ_n = coins.get("in_circulation") or 0
+    spent_n = coins.get("spent") or 0
+    total_n = coins.get("total_coins") or 0
+    amt_alt = coins.get("amount_in_circulation_alt") or coins.get("amount_in_circulation_s") or "?"
+    spent_alt = coins.get("amount_spent_alt") or coins.get("amount_spent_s") or "0"
+    print("  total coins        %s" % total_n)
+    print("  in circulation     %s coins · %s" % (circ_n, amt_alt))
+    print("  spent (in dump)    %s coins · %s" % (spent_n, spent_alt))
+    circ = coins.get("by_denom_circulation") or []
+    if circ:
+        print("  denoms in circ:")
+        for x in circ:
+            print(
+                "    · %s × %s  = %s"
+                % (
+                    x.get("denom_alt") or x.get("denom"),
+                    x.get("count"),
+                    x.get("amount_alt") or x.get("amount"),
+                )
+            )
+    spent = coins.get("by_denom_spent") or []
+    if spent:
+        print("  denoms spent:")
+        for x in spent:
+            print(
+                "    · %s × %s  = %s"
+                % (
+                    x.get("denom_alt") or x.get("denom"),
+                    x.get("count"),
+                    x.get("amount_alt") or x.get("amount"),
+                )
+            )
+else:
+    print("  (no coins-final.json — run with wallet snaps)")
+
+# --- withdrawn / spent flows ---
+print("")
+print("── Money flow (this run) ──────────────────────────────────────")
+# Prefer ladder TSV if provided; else flow-*.txt from e2e
+wd_ok = {"OK", "OK_BANK", "ZERO_REJECT", "ZERO_SKIP", "SKIP_DENOM", "CEILING_REJECT"}
+pay_ok = {"OK", "ZERO_SKIP", "CEILING_SKIP", "CEILING_REJECT"}
+c1, wd_sum, wd_n = sum_tsv_amounts(wd_tsv, "amount", None)
+# count successful withdraws by status prefix OK or soft
+if wd_tsv and Path(wd_tsv).is_file():
+    c1, wd_sum, wd_n = Decimal(0), Decimal(0), 0
+    ccy = cur
+    with open(wd_tsv, newline="") as f:
+        for row in csv.DictReader(f, delimiter="\t"):
+            st = (row.get("status") or "").upper()
+            # count value when mint/settle produced money in wallet
+            if st in ("OK", "OK_BANK"):
+                c, v = parse_amt(row.get("amount"))
+                ccy, wd_sum, wd_n = c, wd_sum + v, wd_n + 1
+    c1 = ccy
+else:
+    c1, wd_sum, wd_n = sum_amount_file(mdir / "flow-withdrawn.txt")
+
+if pay_tsv and Path(pay_tsv).is_file():
+    c2, pay_sum, pay_n = Decimal(0), Decimal(0), 0
+    ccy = cur
+    with open(pay_tsv, newline="") as f:
+        for row in csv.DictReader(f, delimiter="\t"):
+            st = (row.get("status") or "").upper()
+            if st == "OK":
+                c, v = parse_amt(row.get("amount"))
+                ccy, pay_sum, pay_n = c, pay_sum + v, pay_n + 1
+    c2 = ccy
+else:
+    c2, pay_sum, pay_n = sum_amount_file(mdir / "flow-spent.txt")
+
+print("  withdrawn          %s events · %s" % (wd_n, fmt_alt(c1, wd_sum)))
+print("  spent (paid)       %s events · %s" % (pay_n, fmt_alt(c2, pay_sum)))
+net = wd_sum - pay_sum
+print("  net (wd − spent)   %s  %s" % (fmt_alt(c1, net), trend_label(net)))
+if coins.get("ok"):
+    # circulation amount vs net flow
+    circ_map = coins.get("amount_in_circulation") or {}
+    circ_v = Decimal(0)
+    for _c, vs in circ_map.items():
+        circ_v += D(vs)
+    print("  wallet circ now    %s" % fmt_alt(cur, circ_v))
+    print("  residual check     wallet_circ vs net: Δ %s"
+          % fmt_alt(cur, circ_v - net))
+
+# --- coin history tendency ---
+print("")
+print("── Tendency · coins over run ─────────────────────────────────")
+hist = mdir / "coins-history.tsv"
+if hist.is_file() and hist.stat().st_size > 0:
+    rows = list(csv.DictReader(open(hist), delimiter="\t"))
+    if rows:
+        def circ_amt(row):
+            # amount_circ column may be "GOA:20" or alt form — try numeric from total/in_circ
+            a = row.get("amount_circ") or "0"
+            if ":" in a and " " not in a:
+                return parse_amt(a)[1]
+            # try extract GOA: from parentheses
+            import re
+            m = re.search(r"\(([^)]+:[0-9.]+)\)", a)
+            if m:
+                return parse_amt(m.group(1))[1]
+            return D(row.get("in_circ") or 0)
+
+        first, last = rows[0], rows[-1]
+        c0, c1_ = D(first.get("in_circ") or 0), D(last.get("in_circ") or 0)
+        a0, a1 = circ_amt(first), circ_amt(last)
+        print("  samples            %d snaps  (%s → %s)"
+              % (len(rows), first.get("label"), last.get("label")))
+        print("  coins in circ      %s → %s  (Δ %+d)  %s"
+              % (fmt_num(c0), fmt_num(c1_), int(c1_ - c0), trend_label(c1_ - c0)))
+        print("  amount in circ     %s → %s  %s"
+              % (fmt_alt(cur, a0), fmt_alt(cur, a1), trend_label(a1 - a0)))
+        # simple linear slope on coin count
+        if len(rows) >= 3:
+            ys = [float(D(r.get("in_circ") or 0)) for r in rows]
+            n = len(ys)
+            xs = list(range(n))
+            xm, ym = sum(xs) / n, sum(ys) / n
+            num = sum((x - xm) * (y - ym) for x, y in zip(xs, ys))
+            den = sum((x - xm) ** 2 for x in xs) or 1
+            slope = num / den
+            print("  trend slope        %+.3f coins/snap  %s"
+                  % (slope, "↑ accumulating" if slope > 0.05 else ("↓ draining" if slope < -0.05 else "→ flat")))
+        print("  history (label · in_circ · amount):")
+        for r in rows[-12:]:  # last 12
+            print("    · %-28s  circ=%-4s  %s"
+                  % (r.get("label", "?")[:28], r.get("in_circ"), r.get("amount_circ")))
+        if len(rows) > 12:
+            print("    · … (%d earlier snaps)" % (len(rows) - 12))
+    else:
+        print("  (empty history)")
+else:
+    print("  (no coins-history.tsv)")
+
+# --- performance ---
+print("")
+print("── Performance indicators ───────────────────────────────────")
+perf = {}
+pf = mdir / "perf-summary.json"
+if pf.is_file():
+    try:
+        perf = json.load(open(pf))
+    except Exception:
+        perf = {}
+
+def print_bucket(name, v):
+    if not isinstance(v, dict) or not v.get("n"):
+        return
+    print(
+        "  %-14s n=%s  min=%sms  p50=%sms  avg=%sms  max=%sms"
+        % (name, v.get("n"), v.get("min_ms"), v.get("p50_ms"), v.get("avg_ms"), v.get("max_ms"))
+    )
+
+if perf:
+    for k, v in perf.items():
+        print_bucket(k, v)
+else:
+    print("  (no perf-summary.json)")
+
+# per-rung timing tendency from ladder TSV
+for label, path, cols in (
+    ("withdraw rungs", wd_tsv, ("ms_total", "ms_mint", "ms_settle")),
+    ("pay rungs", pay_tsv, ("ms_total", "ms_order", "ms_handle", "ms_settle")),
+):
+    if not path or not Path(path).is_file():
+        continue
+    with open(path, newline="") as f:
+        rrows = list(csv.DictReader(f, delimiter="\t"))
+    if len(rrows) < 2:
+        continue
+    print("  tendency · %s:" % label)
+    for col in cols:
+        vals = []
+        for r in rrows:
+            try:
+                vals.append(int(r.get(col) or 0))
+            except Exception:
+                pass
+        vals = [v for v in vals if v > 0]
+        if len(vals) < 2:
+            continue
+        lab, a, b = half_trend(vals)
+        if a is None:
+            print("    %-12s %s" % (col, lab))
+        else:
+            print("    %-12s first-half avg=%.0fms → second-half avg=%.0fms  %s"
+                  % (col, a, b, lab))
+
+# --- load ---
+print("")
+print("── Host / container load ────────────────────────────────────")
+lb, la = mdir / "load-before.json", mdir / "load-after.json"
+
+def load_brief(path, tag):
+    if not path.is_file():
+        print("  %s: (missing)" % tag)
+        return None
+    try:
+        d = json.load(open(path))
+    except Exception:
+        print("  %s: (unreadable)" % tag)
+        return None
+    if not d.get("ok"):
+        print("  %s: %s" % (tag, d.get("reason", "?")))
+        return d
+    h = d.get("host") or {}
+    mem = h.get("memory") or {}
+    la_ = h.get("loadavg") or []
+    avail = mem.get("mem_available_b")
+    tot = mem.get("mem_total_b")
+    def gi(b):
+        return "?" if b is None else "%.2f GiB" % (b / 1024 / 1024 / 1024)
+    used = (tot - avail) if (tot is not None and avail is not None) else None
+    print("  %s loadavg=%s  mem_used=%s / %s"
+          % (tag, la_, gi(used), gi(tot)))
+    for role in ("bank", "exchange", "merchant"):
+        c = (d.get("taler") or {}).get(role) or {}
+        if not c.get("running"):
+            continue
+        pr = c.get("processes") or {}
+        st = c.get("podman_stats") or {}
+        rss = pr.get("rss_total_b")
+        rss_s = "?" if rss is None else "%.2f GiB" % (rss / 1024 / 1024 / 1024)
+        print("    %-8s rss=%s  cpu=%s  procs=%s"
+              % (role, rss_s, st.get("cpu_pct") or "?", pr.get("proc_total")))
+    return d
+
+db = load_brief(lb, "before")
+da = load_brief(la, "after")
+if db and da and db.get("ok") and da.get("ok"):
+    bla = (db.get("host") or {}).get("loadavg") or [0]
+    ala = (da.get("host") or {}).get("loadavg") or [0]
+    if bla and ala:
+        dload = float(ala[0]) - float(bla[0])
+        print("  loadavg1 tendency  %.2f → %.2f  (Δ %+.2f)  %s"
+              % (float(bla[0]), float(ala[0]), dload,
+                 "↑ higher load" if dload > 0.1 else ("↓ lower load" if dload < -0.1 else "→ steady")))
+    bm = (db.get("host") or {}).get("memory") or {}
+    am = (da.get("host") or {}).get("memory") or {}
+    if bm.get("mem_available_b") is not None and am.get("mem_available_b") is not None:
+        dmem = (am["mem_available_b"] - bm["mem_available_b"]) / 1024 / 1024 / 1024
+        print("  mem_avail tendency %.2f → %.2f GiB  (Δ %+.3f)  %s"
+              % (bm["mem_available_b"] / 1024**3, am["mem_available_b"] / 1024**3, dmem,
+                 "↑ more free" if dmem > 0.05 else ("↓ less free" if dmem < -0.05 else "→ steady")))
+
+print("")
+print("──────────────────────────────────────────────────────────────")
+PY
+}