gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-08-01 09:47:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: manual buy and sell with creator fee

Browse Source
This commit is contained in:
smypmsa
2025-05-13 07:38:46 +00:00
parent 1e34b5b9e9
commit f811632c3c
2 changed files with 90 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
learning-examples/manual_buy.py
+27 -6
View File
@@ -8,7 +8,7 @@ import struct
import base58
import spl.token.instructions as spl_token
import websockets
from construct import Flag, Int64ul, Struct
from construct import Bytes, Flag, Int64ul, Struct
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
from solana.rpc.types import TxOpts
@@ -36,7 +36,6 @@ SYSTEM_TOKEN_PROGRAM = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss62
SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM = Pubkey.from_string(
"ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL"
)
SYSTEM_RENT = Pubkey.from_string("SysvarRent111111111111111111111111111111111")
SOL = Pubkey.from_string("So11111111111111111111111111111111111111112")
LAMPORTS_PER_SOL = 1_000_000_000
@@ -53,11 +52,20 @@ class BondingCurveState:
"real_sol_reserves" / Int64ul,
"token_total_supply" / Int64ul,
"complete" / Flag,
"creator" / Bytes(32), # Added new creator field - 32 bytes for Pubkey
)
def __init__(self, data: bytes) -> None:
"""Parse bonding curve data."""
if data[:8] != EXPECTED_DISCRIMINATOR:
raise ValueError("Invalid curve state discriminator")
parsed = self._STRUCT.parse(data[8:])
self.__dict__.update(parsed)
# Convert raw bytes to Pubkey for creator field
if hasattr(self, 'creator') and isinstance(self.creator, bytes):
self.creator = Pubkey.from_bytes(self.creator)
async def get_pump_curve_state(
@@ -83,10 +91,22 @@ def calculate_pump_curve_price(curve_state: BondingCurveState) -> float:
)
def _find_creator_vault(creator: Pubkey) -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[
b"creator-vault",
bytes(creator)
],
PUMP_PROGRAM,
)
return derived_address
async def buy_token(
mint: Pubkey,
bonding_curve: Pubkey,
associated_bonding_curve: Pubkey,
creator_vault: Pubkey,
amount: float,
slippage: float = 0.25,
max_retries=5,
@@ -188,7 +208,7 @@ async def buy_token(
AccountMeta(
pubkey=SYSTEM_TOKEN_PROGRAM, is_signer=False, is_writable=False
),
AccountMeta(pubkey=SYSTEM_RENT, is_signer=False, is_writable=False),
AccountMeta(pubkey=creator_vault, is_signer=False, is_writable=True),
AccountMeta(
pubkey=PUMP_EVENT_AUTHORITY, is_signer=False, is_writable=False
),
@@ -254,8 +274,8 @@ def decode_create_instruction(ix_data, ix_def, accounts):
offset += 4
value = ix_data[offset : offset + length].decode("utf-8")
offset += length
elif arg["type"] == "publicKey":
value = base64.b64encode(ix_data[offset : offset + 32]).decode("utf-8")
elif arg["type"] == "pubkey":
value = base58.b58encode(ix_data[offset : offset + 32]).decode("utf-8")
offset += 32
else:
raise ValueError(f"Unsupported type: {arg['type']}")
@@ -362,6 +382,7 @@ async def main():
mint = Pubkey.from_string(token_data["mint"])
bonding_curve = Pubkey.from_string(token_data["bondingCurve"])
associated_bonding_curve = Pubkey.from_string(token_data["associatedBondingCurve"])
creator_vault = _find_creator_vault(Pubkey.from_string(token_data["creator"]))
# Fetch the token price
async with AsyncClient(RPC_ENDPOINT) as client:
@@ -377,7 +398,7 @@ async def main():
print(
f"Buying {amount:.6f} SOL worth of the new token with {slippage * 100:.1f}% slippage tolerance..."
)
await buy_token(mint, bonding_curve, associated_bonding_curve, amount, slippage)
await buy_token(mint, bonding_curve, associated_bonding_curve, creator_vault, amount, slippage)
if __name__ == "__main__":
learning-examples/manual_sell.py
+63 -9
View File
@@ -3,7 +3,7 @@ import os
import struct
import base58
from construct import Flag, Int64ul, Struct
from construct import Bytes, Flag, Int64ul, Struct
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
from solana.rpc.types import TxOpts
@@ -49,11 +49,61 @@ class BondingCurveState:
"real_sol_reserves" / Int64ul,
"token_total_supply" / Int64ul,
"complete" / Flag,
"creator" / Bytes(32), # Added new creator field - 32 bytes for Pubkey
)
def __init__(self, data: bytes) -> None:
"""Parse bonding curve data."""
if data[:8] != EXPECTED_DISCRIMINATOR:
raise ValueError("Invalid curve state discriminator")
parsed = self._STRUCT.parse(data[8:])
self.__dict__.update(parsed)
# Convert raw bytes to Pubkey for creator field
if hasattr(self, 'creator') and isinstance(self.creator, bytes):
self.creator = Pubkey.from_bytes(self.creator)
async def get_pump_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 get_bonding_curve_address(mint: Pubkey) -> tuple[Pubkey, int]:
return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], PUMP_PROGRAM)
def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[
bytes(bonding_curve),
bytes(SYSTEM_TOKEN_PROGRAM),
bytes(mint),
],
SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM,
)
return derived_address
def find_creator_vault(creator: Pubkey) -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[
b"creator-vault",
bytes(creator)
],
PUMP_PROGRAM,
)
return derived_address
async def get_pump_curve_state(
@@ -90,6 +140,7 @@ async def sell_token(
mint: Pubkey,
bonding_curve: Pubkey,
associated_bonding_curve: Pubkey,
creator_vault: Pubkey,
slippage: float = 0.25,
max_retries=5,
):
@@ -148,9 +199,9 @@ async def sell_token(
pubkey=SYSTEM_PROGRAM, is_signer=False, is_writable=False
),
AccountMeta(
pubkey=SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM,
pubkey=creator_vault,
is_signer=False,
is_writable=False,
is_writable=True,
),
AccountMeta(
pubkey=SYSTEM_TOKEN_PROGRAM, is_signer=False, is_writable=False
@@ -199,17 +250,20 @@ async def sell_token(
async def main():
# Replace these with the actual values for the token you want to sell
mint = Pubkey.from_string("7aXYndxpkHRU9xg1hyAE6z3X3KYPc2LR3dJMzYTSpump")
bonding_curve = Pubkey.from_string("5UYdCZigGDyAh1doCW8FdnA7VwJTJePayTLCyZkAWMxg")
associated_bonding_curve = Pubkey.from_string(
"BS2RUaPXjQpfzncizvmEfnARZG6SNp1imXnv5USA7PMA"
)
mint = Pubkey.from_string("5GkGpfvRLusWGqxSkwk7uEs6mHzNxH8QaZD6uPvYpump")
bonding_curve, _ = get_bonding_curve_address(mint)
associated_bonding_curve = find_associated_bonding_curve(mint, bonding_curve)
async with AsyncClient(RPC_ENDPOINT) as client:
curve_state = await get_pump_curve_state(client, bonding_curve)
creator_vault = find_creator_vault(curve_state.creator)
slippage = 0.25 # 25% slippage tolerance
print(f"Bonding curve address: {bonding_curve}")
print(f"Selling tokens with {slippage * 100:.1f}% slippage tolerance...")
await sell_token(mint, bonding_curve, associated_bonding_curve, slippage)
await sell_token(mint, bonding_curve, associated_bonding_curve, creator_vault, slippage)
if __name__ == "__main__":