mirror of
https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git
synced 2026-07-27 23:37:45 +00:00
smypmsa
113 lines
3.3 KiB
Python
113 lines
3.3 KiB
Python
import argparse
|
|
import asyncio
|
|
import os
|
|
import struct
|
|
import sys
|
|
from typing import Final
|
|
|
|
from construct import Flag, Int64ul, Struct
|
|
from solana.rpc.async_api import AsyncClient
|
|
from solders.pubkey import Pubkey
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from core.pubkeys import PumpAddresses
|
|
|
|
# Constants
|
|
EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)
|
|
|
|
RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
def __init__(self, data: bytes) -> None:
|
|
parsed = self._STRUCT.parse(data[8:])
|
|
self.__dict__.update(parsed)
|
|
|
|
|
|
def get_associated_bonding_curve_address(
|
|
mint: Pubkey, program_id: Pubkey
|
|
) -> tuple[Pubkey, int]:
|
|
"""
|
|
Derives the associated bonding curve address for a given mint
|
|
"""
|
|
return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)
|
|
|
|
|
|
async def get_bonding_curve_state(
|
|
conn: AsyncClient, curve_address: Pubkey
|
|
) -> BondingCurveState:
|
|
response = await conn.get_account_info(curve_address)
|
|
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:
|
|
try:
|
|
mint = Pubkey.from_string(mint_address)
|
|
|
|
# Get the associated bonding curve address
|
|
bonding_curve_address, bump = get_associated_bonding_curve_address(
|
|
mint, PumpAddresses.PROGRAM
|
|
)
|
|
|
|
print("\nToken Status:")
|
|
print("-" * 50)
|
|
print(f"Token Mint: {mint}")
|
|
print(f"Associated Bonding Curve: {bonding_curve_address}")
|
|
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
|
|
)
|
|
|
|
print("\nBonding Curve Status:")
|
|
print("-" * 50)
|
|
print(
|
|
f"Completion Status: {'Completed' if curve_state.complete else 'Not Completed'}"
|
|
)
|
|
if curve_state.complete:
|
|
print(
|
|
"\nNote: This bonding curve has completed and liquidity has been migrated to Raydium."
|
|
)
|
|
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():
|
|
parser = argparse.ArgumentParser(description="Check token bonding curve status")
|
|
parser.add_argument("mint_address", help="The token mint address")
|
|
|
|
args = parser.parse_args()
|
|
asyncio.run(check_token_status(args.mint_address))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|