Accessing & Working with Public On-Chain Data

"""Exercise 1: take a snapshot of one validated Mainnet ledger.

Run:  python ex1_ledger_snapshot.py
Needs: pip install xrpl-py==5.2.0
"""
import os
from collections import Counter

from xrpl.clients import JsonRpcClient
from xrpl.models.requests import Ledger, ServerInfo

# Public servers are shared and rate limited: be polite, cache what you
# download, and never hammer them in a tight loop.
# Public full-history Mainnet servers: https://s2.ripple.com:51234/ (Ripple) or
# https://xrplcluster.com/ (InFTF). Override with: export XRPL_RPC=<url>
client = JsonRpcClient(os.environ.get("XRPL_RPC", "https://s2.ripple.com:51234/"))

# 1. Which ledger is the latest validated one?
info = client.request(ServerInfo()).result["info"]
latest = info["validated_ledger"]["seq"]
print(f"Latest validated ledger: {latest}")
print(f"Reserves: base {info['validated_ledger']['reserve_base_xrp']} XRP, "
      f"owner {info['validated_ledger']['reserve_inc_xrp']} XRP per object")

# 2. Fetch that ledger with every transaction expanded (full JSON, not just hashes).
ledger = client.request(
    Ledger(ledger_index=latest, transactions=True, expand=True)
).result["ledger"]

txs = ledger["transactions"]
print(f"Ledger {ledger['ledger_index']} closed at {ledger.get('close_time_iso')} "
      f"with {len(txs)} transactions")

# 3. In API v2 each entry holds the signed fields in `tx_json` and the outcome in `meta`.
types = Counter()
results = Counter()
for entry in txs:
    tx = entry.get("tx_json", entry)
    meta = entry.get("meta") or entry.get("metaData") or {}
    types[tx["TransactionType"]] += 1
    results[meta.get("TransactionResult", "?")] += 1

print("\nTransaction types:")
for name, count in types.most_common():
    print(f"  {name:<22} {count}")
print("\nResults:")
for name, count in results.most_common():
    print(f"  {name:<22} {count}")

"""Exercise 2: download an account's history and turn it into a table.

Run:  python ex2_account_history.py rACCOUNT [max_pages] [ledger_window]
Needs: pip install xrpl-py==5.2.0 pandas
"""
import os
import sys
import time

import pandas as pd
from xrpl.clients import JsonRpcClient
from xrpl.models.requests import AccountTx, ServerInfo

# Public full-history Mainnet servers: https://s2.ripple.com:51234/ (Ripple) or
# https://xrplcluster.com/ (InFTF). Override with: export XRPL_RPC=<url>
client = JsonRpcClient(os.environ.get("XRPL_RPC", "https://s2.ripple.com:51234/"))
if len(sys.argv) < 2:
    sys.exit("usage: python ex2_account_history.py rACCOUNT [max_pages] [ledger_window]\n"
             "Tip: pick an active account from the output of ex3_flow_graph.py or from an explorer.")
ACCOUNT = sys.argv[1]
MAX_PAGES = int(sys.argv[2]) if len(sys.argv) > 2 else 5
# About 20,000 ledgers is roughly one day at 3 to 5 seconds per ledger.
LEDGER_WINDOW = int(sys.argv[3]) if len(sys.argv) > 3 else 20_000


def request(req, retries=5):
    """Send a request, backing off and retrying when the public server is busy."""
    for attempt in range(retries):
        response = client.request(req)
        if response.is_successful():
            return response.result
        print(f"  server said {response.result.get('error')}, retrying...")
        time.sleep(2 ** attempt)
    raise RuntimeError(f"request failed: {response.result}")


def parse_amount(amount):
    """Return (value, asset) for any XRPL amount shape."""
    if amount is None or amount == "unavailable":
        return None, None
    if isinstance(amount, str):                      # XRP, expressed in drops
        return int(amount) / 1_000_000, "XRP"
    if "mpt_issuance_id" in amount:                  # Multi-Purpose Token
        return float(amount["value"]), f"MPT:{amount['mpt_issuance_id']}"
    # Issued token: the asset is currency code AND issuer, never the code alone
    return float(amount["value"]), f"{amount['currency']}.{amount['issuer']}"


def to_row(entry):
    tx = entry.get("tx_json", entry.get("tx", {}))
    meta = entry.get("meta", {})
    # delivered_amount is what actually arrived. Never trust Amount / DeliverMax:
    # a partial payment can succeed while delivering far less.
    value, asset = parse_amount(meta.get("delivered_amount"))
    return {
        "hash": entry.get("hash", tx.get("hash")),
        "ledger_index": entry.get("ledger_index", tx.get("ledger_index")),
        "time": entry.get("close_time_iso"),
        "type": tx.get("TransactionType"),
        "result": meta.get("TransactionResult"),
        "account": tx.get("Account"),
        "destination": tx.get("Destination"),
        "destination_tag": tx.get("DestinationTag"),
        "source_tag": tx.get("SourceTag"),
        "delivered_value": value,
        "delivered_asset": asset,
        "fee_xrp": int(tx.get("Fee", 0)) / 1_000_000,
    }


# Pin the range to fixed ledger numbers. A bounded window is cheaper for the server
# (deep history scans burn your rate-limit quota fast) and makes your dataset
# reproducible: the same query returns the same rows tomorrow.
latest = request(ServerInfo())["info"]["validated_ledger"]["seq"]
first = latest - LEDGER_WINDOW
print(f"Fetching {ACCOUNT} from ledger {first} to {latest}")

rows, marker, page = [], None, 0
while page < MAX_PAGES:
    result = request(AccountTx(
        account=ACCOUNT,
        ledger_index_min=first,
        ledger_index_max=latest,
        limit=200,
        forward=False,            # newest first
        marker=marker,
    ))
    rows += [to_row(e) for e in result["transactions"]]
    page += 1
    marker = result.get("marker")
    print(f"page {page}: {len(rows)} transactions so far")
    if marker is None:            # no marker means there is nothing left
        break
    time.sleep(0.5)               # be polite to the shared public server

if not rows:
    sys.exit("No transactions in this window. Try a larger ledger_window or a busier account.")
df = pd.DataFrame(rows)
df["time"] = pd.to_datetime(df["time"])
df.to_csv(f"{ACCOUNT}_history.csv", index=False)

payments = df[(df["type"] == "Payment") & (df["result"] == "tesSUCCESS")]
incoming = payments[payments["destination"] == ACCOUNT]
outgoing = payments[payments["account"] == ACCOUNT]
print(f"\nSaved {len(df)} rows to {ACCOUNT}_history.csv")
print(f"Successful payments: {len(incoming)} in, {len(outgoing)} out")
print(f"Distinct destination tags on incoming payments: "
      f"{incoming['destination_tag'].nunique()}")
print(f"Failed (tec) transactions still recorded on ledger: "
      f"{(df['result'].str.startswith('tec')).sum()}")
print(df[["time", "type", "result", "account", "destination",
          "delivered_value", "delivered_asset"]].head(10).to_string(index=False))

"""Exercise 3: build a value-flow graph from a range of ledgers.

Run:  python ex3_flow_graph.py [number_of_ledgers]
Needs: pip install xrpl-py==5.2.0 pandas networkx
"""
import os
import sys
import time

import networkx as nx
import pandas as pd
from xrpl.clients import JsonRpcClient
from xrpl.models.requests import Ledger, ServerInfo

# Public full-history Mainnet servers: https://s2.ripple.com:51234/ (Ripple) or
# https://xrplcluster.com/ (InFTF). Override with: export XRPL_RPC=<url>
client = JsonRpcClient(os.environ.get("XRPL_RPC", "https://s2.ripple.com:51234/"))
N_LEDGERS = int(sys.argv[1]) if len(sys.argv) > 1 else 50


def request(req, retries=5):
    """Send a request, backing off and retrying when the public server is busy."""
    for attempt in range(retries):
        response = client.request(req)
        if response.is_successful():
            return response.result
        print(f"  server said {response.result.get('error')}, retrying...")
        time.sleep(2 ** attempt)
    raise RuntimeError(f"request failed: {response.result}")


def xrp_delivered(meta):
    """Delivered XRP in whole XRP, or None when the payment delivered a token."""
    amount = meta.get("delivered_amount")
    return int(amount) / 1_000_000 if isinstance(amount, str) else None


latest = request(ServerInfo())["info"]["validated_ledger"]["seq"]
edges = []
for index in range(latest - N_LEDGERS + 1, latest + 1):
    ledger = request(Ledger(ledger_index=index, transactions=True, expand=True))["ledger"]
    for entry in ledger["transactions"]:
        tx = entry.get("tx_json", entry)
        meta = entry.get("meta", {})
        if tx["TransactionType"] != "Payment" or meta.get("TransactionResult") != "tesSUCCESS":
            continue
        value = xrp_delivered(meta)
        if value is None:
            continue                       # keep this first graph to XRP only
        edges.append({
            "src": tx["Account"], "dst": tx["Destination"], "xrp": value,
            "tag": tx.get("DestinationTag"), "ledger": index,
            "time": ledger.get("close_time_iso"), "hash": entry.get("hash"),
        })
    time.sleep(0.2)

df = pd.DataFrame(edges)
df.to_csv("xrp_payments.csv", index=False)
print(f"{len(df)} successful XRP payments across {N_LEDGERS} ledgers")

# Aggregate parallel payments into one weighted, directed edge per pair of accounts.
G = nx.DiGraph()
for (src, dst), group in df.groupby(["src", "dst"]):
    G.add_edge(src, dst, xrp=group["xrp"].sum(), count=len(group))

print(f"Graph: {G.number_of_nodes()} accounts, {G.number_of_edges()} edges")

stats = pd.DataFrame({
    "in_degree": dict(G.in_degree()),
    "out_degree": dict(G.out_degree()),
    "xrp_in": dict(G.in_degree(weight="xrp")),
    "xrp_out": dict(G.out_degree(weight="xrp")),
}).fillna(0)

# A pass-through account receives and forwards almost the same value in the window.
moved = stats[(stats.xrp_in > 0) & (stats.xrp_out > 0)].copy()
moved["pass_through"] = moved[["xrp_in", "xrp_out"]].min(axis=1) / moved[["xrp_in", "xrp_out"]].max(axis=1)

print("\nTop receivers (fan-in):")
print(stats.sort_values("in_degree", ascending=False).head(5).to_string())
print("\nTop senders (fan-out):")
print(stats.sort_values("out_degree", ascending=False).head(5).to_string())
print("\nAccounts forwarding at least 90% of what they received:")
print(moved[moved.pass_through >= 0.9].sort_values("xrp_in", ascending=False).head(5).to_string())