solana_tx_analyzer.py

#!/usr/bin/env python3
"""Analyze a Solana transaction from its signature."""

import json
import subprocess
import struct
import sys
from collections.abc import Callable

# Base58 alphabet (Bitcoin/Solana)
B58_ALPHABET = b"123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def b58decode(s: str) -> bytes:
    """Decode a base58-encoded string."""
    n = 0
    for ch in s.encode():
        n = n * 58 + B58_ALPHABET.index(ch)
    result = n.to_bytes((n.bit_length() + 7) // 8, "big") if n else b""
    pad = len(s) - len(s.lstrip("1"))
    return b"\x00" * pad + result


def compact_u16_size(value: int) -> int:
    """Return the byte size of a compact-u16 encoding."""
    if value <= 0x7F:
        return 1
    if value <= 0x3FFF:
        return 2
    return 3


def estimate_tx_size(tx: dict) -> int:
    """Estimate the serialized transaction size in bytes."""
    sigs = tx["signatures"]
    msg = tx["message"]
    account_keys = msg["accountKeys"]
    instructions = msg["instructions"]
    alt_lookups = msg.get("addressTableLookups", [])
    is_v0 = bool(alt_lookups)

    size = 0
    if is_v0:
        size += 1  # 0x80 version prefix

    # Signatures
    size += compact_u16_size(len(sigs))
    size += 64 * len(sigs)

    # Message header (3 bytes)
    size += 3

    # Account keys
    size += compact_u16_size(len(account_keys))
    size += 32 * len(account_keys)

    # Recent blockhash
    size += 32

    # Instructions
    size += compact_u16_size(len(instructions))
    for ix in instructions:
        size += 1  # program_id_index
        accounts = ix["accounts"]
        size += compact_u16_size(len(accounts))
        size += len(accounts)
        data_bytes = b58decode(ix["data"])
        size += compact_u16_size(len(data_bytes))
        size += len(data_bytes)

    # Address table lookups (v0 only)
    if is_v0:
        size += compact_u16_size(len(alt_lookups))
        for lookup in alt_lookups:
            size += 32  # ALT account key
            writable = lookup.get("writableIndexes", [])
            readonly = lookup.get("readonlyIndexes", [])
            size += compact_u16_size(len(writable))
            size += len(writable)
            size += compact_u16_size(len(readonly))
            size += len(readonly)

    return size


def decode_compute_budget(
    instructions: list, account_keys: list
) -> tuple[int | None, int | None]:
    """Decode ComputeBudget instructions to extract CU limit and price."""
    cu_limit = None
    cu_price_micro_lamports = None

    for ix in instructions:
        program = account_keys[ix["programIdIndex"]]
        if program != "ComputeBudget111111111111111111111111111111":
            continue
        data = b58decode(ix["data"])
        if not data:
            continue
        disc = data[0]
        if disc == 2 and len(data) >= 5:
            cu_limit = struct.unpack_from("<I", data, 1)[0]
        elif disc == 3 and len(data) >= 9:
            cu_price_micro_lamports = struct.unpack_from("<Q", data, 1)[0]

    return cu_limit, cu_price_micro_lamports


def compute_token_deltas(
    pre_token_balances: list, post_token_balances: list, fee_payer: str
) -> tuple[dict, dict]:
    """Compute token balance deltas for the fee payer, grouped by mint."""

    def build_balance_map(balances: list) -> dict:
        result = {}
        for tb in balances:
            if tb.get("owner") == fee_payer:
                key = (tb["accountIndex"], tb["mint"])
                result[key] = int(tb["uiTokenAmount"]["amount"])
        return result

    pre_map = build_balance_map(pre_token_balances)
    post_map = build_balance_map(post_token_balances)

    # Build decimals lookup once from all available balance entries.
    mint_decimals = {}
    for tb in pre_token_balances + post_token_balances:
        mint_decimals.setdefault(tb["mint"], tb["uiTokenAmount"]["decimals"])

    # Aggregate deltas per mint across all account indices.
    mint_deltas: dict[str, int] = {}
    for account_index, mint in set(pre_map) | set(post_map):
        delta = post_map.get((account_index, mint), 0) - pre_map.get((account_index, mint), 0)
        mint_deltas[mint] = mint_deltas.get(mint, 0) + delta

    return mint_deltas, mint_decimals


def find_accounts_by_balance_change(
    all_accounts: list[str],
    pre_balances: list[int],
    post_balances: list[int],
    predicate: Callable[[int, int], bool],
) -> list[str]:
    """Return accounts whose pre/post balances satisfy the given predicate."""
    return [
        all_accounts[i] if i < len(all_accounts) else f"account[{i}]"
        for i, (pre, post) in enumerate(zip(pre_balances, post_balances))
        if predicate(pre, post)
    ]



def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python solana_tx_analyzer.py <transaction_signature>")
        sys.exit(1)

    tx_sig = sys.argv[1]

    result = subprocess.run(
        ["solana", "confirm", "-v", tx_sig, "--output=json"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        print(f"Error fetching transaction: {result.stderr}", file=sys.stderr)
        sys.exit(1)

    data = json.loads(result.stdout)
    tx = data["transaction"]
    meta = data["meta"]
    msg = tx["message"]
    account_keys = msg["accountKeys"]
    slot = data.get("slot")
    block_time = data.get("blockTime")

    # --- Derived values ---

    # Tx size
    tx_size = estimate_tx_size(tx)

    # Instructions
    top_level_ix = len(msg["instructions"])
    inner_ix = sum(len(g["instructions"]) for g in meta.get("innerInstructions", []))

    # Signatures
    num_sigs = len(tx["signatures"])

    # Compute units
    cu_consumed = meta.get("computeUnitsConsumed", 0)
    cu_limit, cu_price_micro = decode_compute_budget(msg["instructions"], account_keys)

    # Max CPI depth (stackHeight 1 = top-level → depth 0)
    all_stack_heights = [ix.get("stackHeight", 1) for ix in msg["instructions"]]
    for group in meta.get("innerInstructions", []):
        for ix in group["instructions"]:
            all_stack_heights.append(ix.get("stackHeight", 1))
    max_cpi_depth = max(all_stack_heights, default=1) - 1

    # Fees
    total_fee_lamports = meta["fee"]
    base_fee_lamports = 5000 * num_sigs
    priority_fee_lamports = total_fee_lamports - base_fee_lamports

    # Accounts (including ALT-loaded)
    loaded = meta.get("loadedAddresses", {})
    loaded_writable = loaded.get("writable", [])
    loaded_readonly = loaded.get("readonly", [])
    all_accounts = account_keys + loaded_writable + loaded_readonly

    pre_bal = meta["preBalances"]
    post_bal = meta["postBalances"]

    accounts_closed = find_accounts_by_balance_change(
        all_accounts, pre_bal, post_bal,
        predicate=lambda pre, post: pre > 0 and post == 0,
    )
    accounts_created = find_accounts_by_balance_change(
        all_accounts, pre_bal, post_bal,
        predicate=lambda pre, post: pre == 0 and post > 0,
    )

    # Token deltas (fee payer's perspective)
    fee_payer = account_keys[0]
    mint_deltas, mint_decimals = compute_token_deltas(
        meta.get("preTokenBalances", []),
        meta.get("postTokenBalances", []),
        fee_payer,
    )

    alt_resolved = len(loaded_writable) + len(loaded_readonly)
    succeeded = meta["err"] is None

    # --- Output ---

    token_deltas = {}
    for mint, delta in mint_deltas.items():
        decimals = mint_decimals.get(mint, 0)
        token_deltas[mint] = {"raw_delta": delta, "decimals": decimals}

    output = {
        "signature": tx_sig,
        "slot": slot,
        "block_time": block_time,
        "succeeded": succeeded,
        "signed_tx_size_bytes": tx_size,
        "num_signatures": num_sigs,
        "num_instructions_total": top_level_ix + inner_ix,
        "num_instructions_top_level": top_level_ix,
        "num_instructions_inner": inner_ix,
        "cu_consumed": cu_consumed,
        "cu_requested": cu_limit,
        "cu_price_micro_lamports": cu_price_micro,
        "max_cpi_depth": max_cpi_depth,
        "base_fee_lamports": base_fee_lamports,
        "priority_fee_lamports": priority_fee_lamports,
        "total_fee_lamports": total_fee_lamports,
        "num_accounts": len(all_accounts),
        "alt_resolved_accounts": alt_resolved,
        "accounts_created": accounts_created,
        "accounts_closed": accounts_closed,
        "token_deltas": token_deltas,
    }

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()