mirror of
https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git
synced 2026-07-28 15:57:43 +00:00
03a4e7bcbc
* 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
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
import base64
|
|
import json
|
|
import struct
|
|
|
|
from construct import Bytes, Flag, Int64ul, Struct
|
|
from solders.pubkey import Pubkey
|
|
|
|
LAMPORTS_PER_SOL = 1_000_000_000
|
|
TOKEN_DECIMALS = 6
|
|
EXPECTED_DISCRIMINATOR = struct.pack("<Q", 6966180631402821399)
|
|
|
|
|
|
class BondingCurveState:
|
|
_STRUCT = Struct(
|
|
"virtual_token_reserves" / Int64ul,
|
|
"virtual_sol_reserves" / Int64ul,
|
|
"real_token_reserves" / Int64ul,
|
|
"real_sol_reserves" / Int64ul,
|
|
"token_total_supply" / Int64ul,
|
|
"complete" / Flag,
|
|
"creator" / Bytes(32), # Added new creator field - 32 bytes for Pubkey
|
|
"is_mayhem_mode" / Flag, # Added mayhem mode flag - 1 byte
|
|
)
|
|
|
|
def __init__(self, data: bytes) -> None:
|
|
"""Parse bonding curve data - supports all versions."""
|
|
if data[:8] != EXPECTED_DISCRIMINATOR:
|
|
raise ValueError("Invalid curve state discriminator")
|
|
|
|
# Required fields (always present)
|
|
offset = 8
|
|
self.virtual_token_reserves = int.from_bytes(
|
|
data[offset : offset + 8], "little"
|
|
)
|
|
offset += 8
|
|
self.virtual_sol_reserves = int.from_bytes(data[offset : offset + 8], "little")
|
|
offset += 8
|
|
self.real_token_reserves = int.from_bytes(data[offset : offset + 8], "little")
|
|
offset += 8
|
|
self.real_sol_reserves = int.from_bytes(data[offset : offset + 8], "little")
|
|
offset += 8
|
|
self.token_total_supply = int.from_bytes(data[offset : offset + 8], "little")
|
|
offset += 8
|
|
self.complete = bool(data[offset])
|
|
offset += 1
|
|
|
|
# Optional fields (may not be present in older versions)
|
|
if len(data) >= offset + 32:
|
|
self.creator = Pubkey.from_bytes(data[offset : offset + 32])
|
|
offset += 32
|
|
|
|
if len(data) > offset:
|
|
self.is_mayhem_mode = bool(data[offset])
|
|
else:
|
|
self.is_mayhem_mode = None
|
|
|
|
else:
|
|
self.creator = None
|
|
|
|
|
|
def calculate_bonding_curve_price(curve_state: BondingCurveState) -> float:
|
|
if curve_state.virtual_token_reserves <= 0 or curve_state.virtual_sol_reserves <= 0:
|
|
raise ValueError("Invalid reserve state")
|
|
|
|
return (curve_state.virtual_sol_reserves / LAMPORTS_PER_SOL) / (
|
|
curve_state.virtual_token_reserves / 10**TOKEN_DECIMALS
|
|
)
|
|
|
|
|
|
def decode_bonding_curve_data(raw_data: str) -> BondingCurveState:
|
|
decoded_data = base64.b64decode(raw_data)
|
|
if decoded_data[:8] != EXPECTED_DISCRIMINATOR:
|
|
raise ValueError("Invalid curve state discriminator")
|
|
return BondingCurveState(decoded_data)
|
|
|
|
|
|
# Load the JSON data
|
|
with open("learning-examples/raw_bondingCurve_from_getAccountInfo.json") as file:
|
|
json_data = json.load(file)
|
|
|
|
# Extract the base64 encoded data
|
|
encoded_data = json_data["result"]["value"]["data"][0]
|
|
|
|
# Decode the data
|
|
bonding_curve_state = decode_bonding_curve_data(encoded_data)
|
|
|
|
# Calculate and print the token price
|
|
token_price_sol = calculate_bonding_curve_price(bonding_curve_state)
|
|
|
|
print("Bonding Curve State:")
|
|
print(f" Virtual Token Reserves: {bonding_curve_state.virtual_token_reserves}")
|
|
print(f" Virtual SOL Reserves: {bonding_curve_state.virtual_sol_reserves}")
|
|
print(f" Real Token Reserves: {bonding_curve_state.real_token_reserves}")
|
|
print(f" Real SOL Reserves: {bonding_curve_state.real_sol_reserves}")
|
|
print(f" Token Total Supply: {bonding_curve_state.token_total_supply}")
|
|
print(f" Complete: {bonding_curve_state.complete}")
|
|
print(f"\nToken Price: {token_price_sol:.10f} SOL")
|