from solana.rpc.api import Client
from solana.rpc.commitment import Commitment
from solders.signature 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()
