Solana: Analyzing adjacent transactions for priority fees

from solana.rpc.api import Client
from solana.rpc.commitment import Commitment
from solana.transaction import Signature
import base58
import json
from typing import List, Optional, Tuple

PROGRAM_ID = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
COMPUTE_BUDGET_ID = "ComputeBudget111111111111111111111111111111"

def get_transaction_slot(client: Client, signature_str: str) -> Optional[int]:
    try:
        signature_bytes = base58.b58decode(signature_str)
        signature = Signature(signature_bytes)

        response = client.get_transaction(
            signature,
            max_supported_transaction_version=0
        )
        if hasattr(response, 'value') and response.value:
            return response.value.slot
        return None
    except Exception as e:
        print(f"Error getting transaction slot: {e}")
        return None

def calculate_transaction_fees(tx) -> Tuple[float, float]:

    total_fee_lamports = tx.meta.fee
    total_fee_sol = total_fee_lamports / 1_000_000_000

    cu_price_micro_lamports = None
    cu_limit = None


    for ix in tx.transaction.message.instructions:
        program_id = tx.transaction.message.account_keys[ix.program_id_index]
        if str(program_id) == COMPUTE_BUDGET_ID:
            data = base58.b58decode(ix.data)
            if len(data) >= 9 and data[0] == 3:
                cu_price_micro_lamports = int.from_bytes(data[1:9], 'little')
            elif len(data) >= 4 and data[0] == 2:
                cu_limit = int.from_bytes(data[1:4], 'little')


    if cu_price_micro_lamports is not None and cu_limit is not None:
        priority_fee_lamports = (cu_price_micro_lamports * cu_limit) // 1_000_000
        priority_fee_sol = priority_fee_lamports / 1_000_000_000
    else:
        priority_fee_sol = 0

    return total_fee_sol, priority_fee_sol

def get_block_transactions(client: Client, slot: int, target_signature: str) -> List[tuple]:
    try:
        response = client.get_block(slot, max_supported_transaction_version=0)
        if not response.value:
            return []

        transactions = []
        for tx in response.value.transactions:

            if any(str(account) == PROGRAM_ID for account in tx.transaction.message.account_keys):
                signature = str(tx.transaction.signatures[0])
                total_fee, priority_fee = calculate_transaction_fees(tx)

                writable_accounts = []
                for idx, account in enumerate(tx.transaction.message.account_keys):

                    if idx < tx.transaction.message.header.num_required_signatures:
                        writable_accounts.append(str(account))


                compute_units = tx.meta.compute_units_consumed if hasattr(tx.meta, 'compute_units_consumed') else None

                tx_info = {
                    'signature': signature,
                    'total_fee': total_fee,
                    'priority_fee': priority_fee,
                    'block': slot,
                    'writable_accounts': writable_accounts,
                    'compute_units': compute_units
                }

                if signature == target_signature:
                    transactions.insert(0, tx_info)
                else:
                    transactions.append(tx_info)

        return transactions
    except Exception as e:
        print(f"Error getting block transactions for slot {slot}: {e}")
        return []

def analyze_slots(rpc_url: str, signature: str, slots_back: int, slots_forward: int):
    client = Client(rpc_url)
    current_slot = get_transaction_slot(client, signature)
    if not current_slot:
        print("Could not find transaction slot")
        return

    print(f"\nAnalyzing slots {current_slot-slots_back} to {current_slot+slots_forward}")
    print("=" * 80)

    for slot in range(current_slot - slots_back, current_slot + slots_forward + 1):
        transactions = get_block_transactions(client, slot, signature)

        print(f"\nSlot {slot}:")
        print("-" * 40)

        if transactions:
            target_tx = None
            other_txs = []

            for tx in transactions:
                if tx['signature'] == signature:
                    target_tx = tx
                else:
                    other_txs.append(tx)


            other_txs.sort(key=lambda x: x['priority_fee'], reverse=True)

            if target_tx:
                print("\n" + "⭐️" * 40)
                print("🎯 TARGET TRANSACTION FOUND 🎯")
                print("⭐️" * 40)
                print(f"\nSignature: {target_tx['signature']}")
                print(f"Block: {target_tx['block']}")
                print(f"{'Total fee:':<15} {target_tx['total_fee']:>12.8f} SOL")
                print(f"{'Priority fee:':<15} {target_tx['priority_fee']:>12.8f} SOL")
                print(f"{'Base fee:':<15} {(target_tx['total_fee'] - target_tx['priority_fee']):>12.8f} SOL")
                print(f"Compute Units: {target_tx['compute_units']}")
                print("\nWritable Accounts:")
                for account in target_tx['writable_accounts']:
                    print(f"  {account}")
                print("⭐️" * 40 + "\n")

            if other_txs:
                print(f"Other transactions in slot {slot} (sorted by priority fee):")
                print("-" * 40)

                for tx in other_txs:
                    print(f"\nSignature: {tx['signature']}")
                    print(f"Block: {tx['block']}")
                    print(f"Fee: {tx['total_fee']:.8f} SOL (Priority: {tx['priority_fee']:.8f} SOL)")
                    print(f"Compute Units: {tx['compute_units']}")
                    print("Writable Accounts:")
                    for account in tx['writable_accounts']:
                        print(f"  {account}")
                    print("-" * 40)
        else:
            print("No relevant transactions")

def main():
    RPC_URL = "CHAINSTACK_NODE_RPC"
    SIGNATURE = "TRANSACTION_SIGNATURE"
    SLOTS_BACK = 1  # Number of slots to check back
    SLOTS_FORWARD = 1  # Number of slots to check forward

    analyze_slots(RPC_URL, SIGNATURE, SLOTS_BACK, SLOTS_FORWARD)

if __name__ == "__main__":
    main()

Slot 309220412:
----------------------------------------

⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️
🎯 TARGET TRANSACTION FOUND 🎯
⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️

Signature: 3Bd4wJ3pZrPYbhnHpUbZFXfunD2A1dciNx8umhhTEg4KLYmP1nERNbKEbHP9j6K8CcJgdekRi3PhUqVjdsHcsMSA
Block: 309220412
Total fee:        0.00200500 SOL
Priority fee:     0.00200000 SOL
Base fee:         0.00000500 SOL
Compute Units: 57900

Writable Accounts:
  8SDfmRCMwn4qNvBpnvMTX7hunGHXWEJ5odhazNVcFyBw
⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️⭐️

Other transactions in slot 309220412 (sorted by priority fee):
----------------------------------------