mirror of
https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git
synced 2026-07-28 15:57:43 +00:00
Anton
d64b51da44
* fix(examples): remove hardcoded token mints * feat(core): update pump idls * feat(examples): add track volume bool to mint script * feat(examples): add extend acc instr to mint
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
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__), "..")))
|
|
|
|
LAMPORTS_PER_SOL: Final[int] = 1_000_000_000
|
|
TOKEN_DECIMALS: Final[int] = 6
|
|
CURVE_ADDRESS: Final[str] = "YOUR_BONDING_CURVE_ADDRESS_HERE" # Replace with actual bonding curve address
|
|
|
|
# Here and later all the discriminators are precalculated. See learning-examples/calculate_discriminator.py
|
|
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)
|
|
|
|
|
|
async def get_bonding_curve_state(
|
|
conn: AsyncClient, curve_address: Pubkey
|
|
) -> BondingCurveState:
|
|
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)
|
|
|
|
|
|
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
|
|
)
|
|
|
|
|
|
async def main() -> None:
|
|
try:
|
|
async with AsyncClient(RPC_ENDPOINT) as conn:
|
|
curve_address = Pubkey.from_string(CURVE_ADDRESS)
|
|
bonding_curve_state = await get_bonding_curve_state(conn, curve_address)
|
|
token_price_sol = calculate_bonding_curve_price(bonding_curve_state)
|
|
|
|
print("Token price:")
|
|
print(f" {token_price_sol:.10f} SOL")
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
except Exception as e:
|
|
print(f"An unexpected error occurred: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|