Files
pumpfun-bonkfun-bot_github/learning-examples/decode_from_getTransaction.py
T
Anton Sauchyk 03a4e7bcbc Add mayhem mode support and Token2022 integration (#149)
* feat: mayhem update in idl

* feat(examples): update bonding curve scripts

* feat(example): update listen_blocksubscribe

* feat(examples): update geyser listener

* feat(examples): update all new token listeners

* feat(examples): add comments, fix printing, formatting

* feat(examples): pumpswap buy and sell update with mayhem mode

* fix(examples): sell pump amm fee recipient

* feat(examples): update decode scripts

* feat(examples): update fetch price

* feat(examples): buy and sell bonding curve scripts

* feat(examples): add mint with mayhem mode enabled

* feat(examples): improve listening to wallet txs

* feat(examples): migration listener improvements

* feat(examples): global vol accumulator is not writable

* feat(examples): support token/token2022 programs in buy instructions

* feat(examples): token/token2022 for pumpswap buy

* feat(examples): token/token2022 supprot for sell instructions

* feat(bot): support create_v2 with token2022, mayhem mode, other fixes

* fix(bot): support only token2022 in logs and pumportal listeners

* feat(bot): token2022 support in cleanup flow

* fix(bot): update token program handling and improve price validation in trading logic

* feat(bot): enhance token program handling for LetsBonk integration
2025-11-18 13:09:37 +01:00

142 lines
4.3 KiB
Python

import json
import struct
import sys
import base58
tx_file_path = ""
if len(sys.argv) != 2:
tx_file_path = "learning-examples/raw_buy_tx_from_getTransaction.json"
print(f"No path provided, using the path: {tx_file_path}")
else:
tx_file_path = sys.argv[1]
# Load the IDL
with open("idl/pump_fun_idl.json") as f:
idl = json.load(f)
# Load the transaction log
with open(tx_file_path) as f:
tx_log = json.load(f)
# Extract the transaction data
tx_data = tx_log["result"]["transaction"]
print(json.dumps(tx_data, indent=2))
def decode_create_instruction(data):
"""Decode legacy Create instruction (Metaplex tokens)."""
# The Create instruction has 3 string arguments: name, symbol, uri
offset = 8 # Skip the 8-byte discriminator
results = []
for _ in range(3):
length = struct.unpack_from("<I", data, offset)[0]
offset += 4
string_data = data[offset : offset + length].decode("utf-8")
results.append(string_data)
offset += length
return {
"name": results[0],
"symbol": results[1],
"uri": results[2],
"token_standard": "legacy",
"is_mayhem_mode": False,
}
def decode_create_v2_instruction(data):
"""Decode CreateV2 instruction (Token2022 tokens)."""
# The CreateV2 instruction has 3 string arguments: name, symbol, uri + is_mayhem_mode
offset = 8 # Skip the 8-byte discriminator
results = []
for _ in range(3):
length = struct.unpack_from("<I", data, offset)[0]
offset += 4
string_data = data[offset : offset + length].decode("utf-8")
results.append(string_data)
offset += length
# Skip creator pubkey (32 bytes)
offset += 32
# Parse is_mayhem_mode (OptionBool at the end)
is_mayhem_mode = False
if offset < len(data):
is_mayhem_mode = bool(data[offset])
return {
"name": results[0],
"symbol": results[1],
"uri": results[2],
"token_standard": "token2022",
"is_mayhem_mode": is_mayhem_mode,
}
def decode_buy_instruction(data):
# Assuming the buy instruction has a u64 argument for amount
amount = struct.unpack_from("<Q", data, 8)[0]
return {"amount": amount}
def decode_instruction_data(instruction, accounts, data):
if instruction["name"] == "create":
return decode_create_instruction(data)
elif instruction["name"] == "createV2":
return decode_create_v2_instruction(data)
elif instruction["name"] == "buy":
return decode_buy_instruction(data)
else:
return f"Unhandled instruction type: {instruction['name']}"
def find_matching_instruction(accounts, data):
if "instructions" not in idl:
print("Warning: No instructions found in IDL")
return None
for instruction in idl["instructions"]:
if len(instruction["accounts"]) == len(accounts):
return instruction
return None
# Parse the transaction
tx_message = tx_data["message"]
instructions = tx_message["instructions"]
for ix in instructions:
program_id = ix.get("programId")
accounts = ix.get("accounts", [])
data = ix.get("data", "")
if "parsed" in ix:
print(f"Parsed instruction: {ix['program']} - {ix['parsed']['type']}")
print(f"Info: {json.dumps(ix['parsed']['info'], indent=2)}")
elif (
program_id == "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
): # Pump Fun Program
matching_instruction = find_matching_instruction(accounts, data)
if matching_instruction:
decoded_data = decode_instruction_data(
matching_instruction, accounts, base58.b58decode(data)
)
print(f"Instruction: {matching_instruction['name']}")
print(f"Decoded data: {decoded_data}")
print("\nAccounts:")
for i, account in enumerate(accounts):
account_info = matching_instruction["accounts"][i]
print(f" {account_info['name']}: {account}")
else:
print(f"Unable to match instruction for program {program_id}")
else:
print(f"Instruction for program: {program_id}")
print(f"Data: {data}\n")
print("\nTransaction Information:")
print(f"Blockhash: {tx_message['recentBlockhash']}")
print(f"Fee payer: {tx_message['accountKeys'][0]['pubkey']}")
print(f"Signature: {tx_data['signatures'][0]}")