pumpfun-bonkfun-bot_github/learning-examples/bonding-curve-progress/get_bonding_curve_status.py

"""
Module for checking the status of a token's bonding curve on the Solana network using
the Pump.fun program. It allows querying the bonding curve state and completion status.
"""

import argparse
import asyncio
import os
import struct
from typing import Final

from construct import Bytes, Flag, Int64ul, Struct
from dotenv import load_dotenv
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey

load_dotenv()

RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")

# Change to token you want to query
TOKEN_MINT = "..."

# Constants
PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string(
    "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
)
EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)

# Shortest account that still reaches quote_mint: 8 discriminator + 41 base fields
# + 32 creator + 1 mayhem + 1 cashback + 32 quote_mint.
_V5_MIN_LENGTH: Final[int] = 115

# Quote assets. `quote_mint` is all zeros on SOL-paired coins; the quote-side reserves
# are always in the quote mint's raw units (1e9 for SOL, 1e6 for USDC).
DEFAULT_QUOTE_MINT: Final[Pubkey] = Pubkey.from_bytes(bytes(32))
WSOL_MINT: Final[Pubkey] = Pubkey.from_string(
    "So11111111111111111111111111111111111111112"
)
USDC_MINT: Final[Pubkey] = Pubkey.from_string(
    "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
)
QUOTE_DECIMALS: Final[dict[Pubkey, int]] = {WSOL_MINT: 9, USDC_MINT: 6}
QUOTE_SYMBOLS: Final[dict[Pubkey, str]] = {WSOL_MINT: "SOL", USDC_MINT: "USDC"}


def resolve_quote_asset(quote_mint: Pubkey) -> tuple[Pubkey, str, int]:
    """Resolve a curve's quote mint to its symbol and raw-unit scale.

    Args:
        quote_mint: The curve's raw quote_mint field

    Returns:
        (effective mint, display symbol, raw units per whole token)
    """
    mint = WSOL_MINT if quote_mint == DEFAULT_QUOTE_MINT else quote_mint
    decimals = QUOTE_DECIMALS.get(mint, 9)
    return mint, QUOTE_SYMBOLS.get(mint, str(mint)), 10**decimals


class BondingCurveState:
    """
    Represents the state of a bonding curve account.

    Attributes:
        virtual_token_reserves: Virtual token reserves in the curve
        virtual_sol_reserves: Virtual SOL reserves in the curve
        real_token_reserves: Real token reserves in the curve
        real_sol_reserves: Real SOL reserves in the curve
        token_total_supply: Total token supply in the curve
        complete: Whether the curve has completed and liquidity migrated
        is_mayhem_mode: Whether the curve is in mayhem mode
    """

    # V2: Struct with creator field (81 bytes total: 8 discriminator + 73 data)
    _STRUCT_V2 = 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
    )

    # V3: Struct with creator + mayhem mode (82 bytes total: 8 discriminator + 74 data)
    _STRUCT_V3 = 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),
        "is_mayhem_mode" / Flag,  # Added mayhem mode flag - 1 byte
    )

    # V4: V3 + is_cashback_coin (83 bytes total: 8 discriminator + 75 data) — added in the late-Feb 2026 cashback upgrade
    _STRUCT_V4 = 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),
        "is_mayhem_mode" / Flag,
        "is_cashback_coin" / Flag,
    )

    # V5: V4 + quote_mint. Live accounts are 151 bytes — this struct covers 107 of the
    # 143 data bytes and the remaining 36 are reserved padding. The quote-side reserves
    # are in the quote mint's raw units, so a non-SOL coin must not be scaled by 1e9.
    _STRUCT_V5 = 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),
        "is_mayhem_mode" / Flag,
        "is_cashback_coin" / Flag,
        "quote_mint" / Bytes(32),
    )

    def __init__(self, data: bytes) -> None:
        """Parse bonding curve data."""
        if data[:8] != EXPECTED_DISCRIMINATOR:
            raise ValueError("Invalid curve state discriminator")

        total_length = len(data)
        self.quote_mint = DEFAULT_QUOTE_MINT

        if total_length == 81:  # V2: Creator only
            parsed = self._STRUCT_V2.parse(data[8:])
            self.__dict__.update(parsed)
            self.creator = Pubkey.from_bytes(self.creator)
            self.is_mayhem_mode = False
            self.is_cashback_coin = False

        elif total_length == 82:  # V3: Creator + mayhem
            parsed = self._STRUCT_V3.parse(data[8:])
            self.__dict__.update(parsed)
            self.creator = Pubkey.from_bytes(self.creator)
            self.is_cashback_coin = False

        elif total_length < _V5_MIN_LENGTH:  # V4: Creator + mayhem + cashback
            parsed = self._STRUCT_V4.parse(data[8:])
            self.__dict__.update(parsed)
            self.creator = Pubkey.from_bytes(self.creator)

        elif total_length >= _V5_MIN_LENGTH:  # V5: + quote_mint
            parsed = self._STRUCT_V5.parse(data[8:])
            self.__dict__.update(parsed)
            self.creator = Pubkey.from_bytes(self.creator)
            self.quote_mint = Pubkey.from_bytes(self.quote_mint)

        else:
            raise ValueError(f"Unexpected bonding curve size: {total_length} bytes")


def get_bonding_curve_address(mint: Pubkey, program_id: Pubkey) -> tuple[Pubkey, int]:
    """
    Derives the associated bonding curve address for a given mint.

    Args:
        mint: The token mint address
        program_id: The program ID for the bonding curve

    Returns:
        Tuple of (bonding curve address, bump seed)
    """
    return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)


async def get_bonding_curve_state(
    conn: AsyncClient, curve_address: Pubkey
) -> BondingCurveState:
    """
    Fetches and validates the state of a bonding curve account.

    Args:
        conn: AsyncClient connection to Solana RPC
        curve_address: Address of the bonding curve account

    Returns:
        BondingCurveState object containing parsed account data

    Raises:
        ValueError: If account data is invalid or missing
    """
    response = await conn.get_account_info(curve_address, encoding="base64")
    if not response.value or not response.value.data:
        raise ValueError("Invalid curve state: No data")

    data = response.value.data
    if data[:8] != EXPECTED_DISCRIMINATOR:
        raise ValueError("Invalid curve state discriminator")

    return BondingCurveState(data)


async def check_token_status(mint_address: str) -> None:
    """
    Checks and prints the status of a token and its bonding curve.

    Args:
        mint_address: The token mint address as a string
    """
    try:
        mint = Pubkey.from_string(mint_address)
        bonding_curve_address, bump = get_bonding_curve_address(mint, PUMP_PROGRAM_ID)

        print("\nToken status:")
        print("-" * 50)
        print(f"Token mint:              {mint}")
        print(f"Bonding curve:           {bonding_curve_address}")
        if bump is not None:
            print(f"Bump seed:               {bump}")
        print("-" * 50)

        # Check completion status
        async with AsyncClient(RPC_ENDPOINT) as client:
            try:
                curve_state = await get_bonding_curve_state(
                    client, bonding_curve_address
                )

                quote_mint, quote_symbol, quote_unit = resolve_quote_asset(
                    curve_state.quote_mint
                )

                print("\nBonding curve status:")
                print("-" * 50)
                print(f"Creator:             {curve_state.creator}")
                print(f"Quote asset:         {quote_symbol} ({quote_mint})")
                print(
                    f"Mayhem Mode:         {'✅ Enabled' if curve_state.is_mayhem_mode else '❌ Disabled'}"
                )
                print(
                    f"Cashback Coin:       {'✅ Enabled' if curve_state.is_cashback_coin else '❌ Disabled'}"
                )
                print(
                    f"Completed:           {'✅ Migrated' if curve_state.complete else '❌ Bonding curve'}"
                )

                print("\nBonding curve reserves:")
                print(f"Virtual Token:       {curve_state.virtual_token_reserves:,}")
                print(
                    f"Virtual quote:       {curve_state.virtual_sol_reserves:,} raw "
                    f"({curve_state.virtual_sol_reserves / quote_unit:,.6f} {quote_symbol})"
                )
                print(f"Real Token:          {curve_state.real_token_reserves:,}")
                print(
                    f"Real quote:          {curve_state.real_sol_reserves:,} raw "
                    f"({curve_state.real_sol_reserves / quote_unit:,.6f} {quote_symbol})"
                )
                print(f"Total Supply:        {curve_state.token_total_supply:,}")

                if curve_state.complete:
                    print(
                        "\nNote: This bonding curve has completed and liquidity has been migrated to PumpSwap."
                    )
                print("-" * 50)

            except ValueError as e:
                print(f"\nError accessing bonding curve: {e}")

    except ValueError as e:
        print(f"\nError: Invalid address format - {e}")
    except Exception as e:
        print(f"\nUnexpected error: {e}")


def main() -> None:
    """Main entry point for the token status checker."""
    parser = argparse.ArgumentParser(description="Check token bonding curve status")
    parser.add_argument(
        "mint_address", nargs="?", help="The token mint address", default=TOKEN_MINT
    )
    args = parser.parse_args()

    asyncio.run(check_token_status(args.mint_address))


if __name__ == "__main__":
    main()