gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-07-27 15:27:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: align learning examples with refactored modules

Browse Source
This commit is contained in:
smypmsa
2025-03-19 16:01:46 +00:00
parent e6236a9a86
commit ca91c879c3
14 changed files with 141 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
learning-examples/blockSubscribe_extract_transactions.py
+9 -6
View File
@@ -7,7 +7,10 @@ import sys
import websockets
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from config import PUMP_PROGRAM, WSS_ENDPOINT
from core.pubkeys import PumpAddresses
WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
async def save_transaction(tx_data, tx_signature):
@@ -27,7 +30,7 @@ async def listen_for_transactions():
"id": 1,
"method": "blockSubscribe",
"params": [
{"mentionsAccountOrProgram": str(PUMP_PROGRAM)},
{"mentionsAccountOrProgram": str(PumpAddresses.PROGRAM)},
{
"commitment": "confirmed",
"encoding": "base64",
@@ -36,10 +39,10 @@ async def listen_for_transactions():
"maxSupportedTransactionVersion": 0,
},
],
}
},
)
await websocket.send(subscription_message)
print(f"Subscribed to blocks mentioning program: {PUMP_PROGRAM}")
print(f"Subscribed to blocks mentioning program: {PumpAddresses.PROGRAM}")
while True:
try:
@@ -71,9 +74,9 @@ async def listen_for_transactions():
continue
await save_transaction(tx, tx_signature)
elif "result" in data:
print(f"Subscription confirmed")
print("Subscription confirmed")
except Exception as e:
print(f"An error occurred: {str(e)}")
print(f"An error occurred: {e!s}")
if __name__ == "__main__":
learning-examples/check_boding_curve_status.py
+5 -2
View File
@@ -10,11 +10,14 @@ 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 config import PUMP_PROGRAM, RPC_ENDPOINT
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(
@@ -60,7 +63,7 @@ async def check_token_status(mint_address: str) -> None:
# Get the associated bonding curve address
bonding_curve_address, bump = get_associated_bonding_curve_address(
mint, PUMP_PROGRAM
mint, PumpAddresses.PROGRAM
)
print("\nToken Status:")
learning-examples/compute_associated_bonding_curve.py
+7 -6
View File
@@ -4,7 +4,8 @@ import sys
from solders.pubkey import Pubkey
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from config import PUMP_PROGRAM
from core.pubkeys import PumpAddresses, SystemAddresses
def get_bonding_curve_address(mint: Pubkey, program_id: Pubkey) -> tuple[Pubkey, int]:
@@ -19,16 +20,14 @@ def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey
Find the associated bonding curve for a given mint and bonding curve.
This uses the standard ATA derivation.
"""
from config import SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM as ATA_PROGRAM_ID
from config import SYSTEM_TOKEN_PROGRAM as TOKEN_PROGRAM_ID
derived_address, _ = Pubkey.find_program_address(
[
bytes(bonding_curve),
bytes(TOKEN_PROGRAM_ID),
bytes(SystemAddresses.TOKEN_PROGRAM),
bytes(mint),
],
ATA_PROGRAM_ID,
SystemAddresses.ASSOCIATED_TOKEN_PROGRAM,
)
return derived_address
@@ -39,7 +38,9 @@ def main():
try:
mint = Pubkey.from_string(mint_address)
bonding_curve_address, bump = get_bonding_curve_address(mint, PUMP_PROGRAM)
bonding_curve_address, bump = get_bonding_curve_address(
mint, PumpAddresses.PROGRAM
)
# Calculate the associated bonding curve
associated_bonding_curve = find_associated_bonding_curve(
learning-examples/decode_from_blockSubscribe.py
+13 -12
View File
@@ -4,18 +4,16 @@ import json
import struct
import sys
from solana.transaction import Transaction
from solders.pubkey import Pubkey
from solders.transaction import VersionedTransaction
from solders.transaction import Transaction, VersionedTransaction
def load_idl(file_path):
with open(file_path, "r") as f:
with open(file_path) as f:
return json.load(f)
def load_transaction(file_path):
with open(file_path, "r") as f:
with open(file_path) as f:
data = json.load(f)
return data
@@ -67,8 +65,8 @@ def decode_transaction(tx_data, idl):
print("Versioned transaction detected")
else:
# Use legacy deserialization for older transactions
transaction = Transaction.deserialize(tx_data_decoded)
instructions = transaction.instructions
transaction = Transaction.from_bytes(tx_data_decoded)
instructions = transaction.message.instructions
account_keys = transaction.message.account_keys
print("Legacy transaction detected")
@@ -137,12 +135,15 @@ def decode_transaction(tx_data, idl):
return decoded_instructions
if len(sys.argv) != 2:
print("Usage: python decode_fromBlock.py <transaction_file_path>")
sys.exit(1)
tx_file_path = ""
tx_file_path = sys.argv[1]
idl = load_idl("../idl/pump_fun_idl.json")
if len(sys.argv) != 2:
tx_file_path = "learning-examples/blockSubscribe-transactions/raw_create_tx_from_blockSubscribe.json"
print(f"No path provided, using the path: {tx_file_path}")
else:
tx_file_path = sys.argv[1]
idl = load_idl("idl/pump_fun_idl.json")
tx_data = load_transaction(tx_file_path)
decoded_instructions = decode_transaction(tx_data, idl)
learning-examples/decode_from_getAccountInfo.py
+1 -1
View File
@@ -41,7 +41,7 @@ def decode_bonding_curve_data(raw_data: str) -> BondingCurveState:
# Load the JSON data
with open("raw_bondingCurve_from_getAccountInfo.json", "r") as file:
with open("learning-examples/raw_bondingCurve_from_getAccountInfo.json", "r") as file:
json_data = json.load(file)
# Extract the base64 encoded data
learning-examples/decode_from_getTransaction.py
+8 -9
View File
@@ -1,24 +1,23 @@
import base64
import json
import struct
import sys
import base58
from solana.transaction import Transaction
from solders.pubkey import Pubkey
tx_file_path = ""
if len(sys.argv) != 2:
print("Usage: python decode_getTransaction.py <transaction_file_path>")
sys.exit(1)
tx_file_path = sys.argv[1]
tx_file_path = "learning-examples/raw_buy_tx_from_getTransaction.json"
print(f"No path provided, using the path: {tx_file_path}")
else:
tx_file_path = sys.argv[1]
# Load the IDL
with open("../idl/pump_fun_idl.json", "r") as f:
with open("idl/pump_fun_idl.json") as f:
idl = json.load(f)
# Load the transaction log
with open(tx_file_path, "r") as f:
with open(tx_file_path) as f:
tx_log = json.load(f)
# Extract the transaction data
learning-examples/fetch_price.py
+2 -1
View File
@@ -9,7 +9,6 @@ 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 config import RPC_ENDPOINT
LAMPORTS_PER_SOL: Final[int] = 1_000_000_000
TOKEN_DECIMALS: Final[int] = 6
@@ -18,6 +17,8 @@ CURVE_ADDRESS: Final[str] = "6GXfUqrmPM4VdN1NoDZsE155jzRegJngZRjMkGyby7do"
# Here and later all the discriminators are precalculated. See learning-examples/discriminator.py
EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)
RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
class BondingCurveState:
_STRUCT = Struct(
learning-examples/listen_create_from_blocksubscribe.py
+10 -10
View File
@@ -1,21 +1,21 @@
import asyncio
import base64
import hashlib
import json
import os
import struct
import sys
import websockets
from solders.pubkey import Pubkey
from solders.transaction import VersionedTransaction
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from config import PUMP_PROGRAM, WSS_ENDPOINT
from core.pubkeys import PumpAddresses
WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
def load_idl(file_path):
with open(file_path, "r") as f:
with open(file_path) as f:
return json.load(f)
@@ -48,7 +48,7 @@ def decode_create_instruction(ix_data, ix_def, accounts):
# Here and later all the discriminators are precalculated. See learning-examples/discriminator.py
async def listen_and_decode_create():
idl = load_idl("../idl/pump_fun_idl.json")
idl = load_idl("idl/pump_fun_idl.json")
create_discriminator = 8576854823835016728
async with websockets.connect(WSS_ENDPOINT) as websocket:
@@ -58,7 +58,7 @@ async def listen_and_decode_create():
"id": 1,
"method": "blockSubscribe",
"params": [
{"mentionsAccountOrProgram": str(PUMP_PROGRAM)},
{"mentionsAccountOrProgram": str(PumpAddresses.PROGRAM)},
{
"commitment": "confirmed",
"encoding": "base64",
@@ -70,7 +70,7 @@ async def listen_and_decode_create():
}
)
await websocket.send(subscription_message)
print(f"Subscribed to blocks mentioning program: {PUMP_PROGRAM}")
print(f"Subscribed to blocks mentioning program: {PumpAddresses.PROGRAM}")
while True:
try:
@@ -97,7 +97,7 @@ async def listen_and_decode_create():
transaction.message.account_keys[
ix.program_id_index
]
) == str(PUMP_PROGRAM):
) == str(PumpAddresses.PROGRAM):
ix_data = bytes(ix.data)
discriminator = struct.unpack(
"<Q", ix_data[:8]
@@ -134,13 +134,13 @@ async def listen_and_decode_create():
)
print("--------------------")
elif "result" in data:
print(f"Subscription confirmed")
print("Subscription confirmed")
else:
print(
f"Received unexpected message type: {data.get('method', 'Unknown')}"
)
except Exception as e:
print(f"An error occurred: {str(e)}")
print(f"An error occurred: {e!s}")
print(f"Error details: {type(e).__name__}")
import traceback
learning-examples/listen_new_direct.py
+10 -5
View File
@@ -9,10 +9,13 @@ import base58
import websockets
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from config import PUMP_PROGRAM, WSS_ENDPOINT
from core.pubkeys import PumpAddresses
WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
# Load the IDL JSON file
with open("../idl/pump_fun_idl.json", "r") as f:
with open("idl/pump_fun_idl.json") as f:
idl = json.load(f)
# Extract the "create" instruction definition
@@ -77,13 +80,15 @@ async def listen_for_new_tokens():
"id": 1,
"method": "logsSubscribe",
"params": [
{"mentions": [str(PUMP_PROGRAM)]},
{"mentions": [str(PumpAddresses.PROGRAM)]},
{"commitment": "processed"},
],
}
)
await websocket.send(subscription_message)
print(f"Listening for new token creations from program: {PUMP_PROGRAM}")
print(
f"Listening for new token creations from program: {PumpAddresses.PROGRAM}"
)
# Wait for subscription confirmation
response = await websocket.recv()
@@ -124,7 +129,7 @@ async def listen_for_new_tokens():
)
except Exception as e:
print(f"Failed to decode: {log}")
print(f"Error: {str(e)}")
print(f"Error: {e!s}")
except Exception as e:
print(f"An error occurred while processing message: {e}")
learning-examples/listen_new_direct_full_details.py
+11 -14
View File
@@ -10,14 +10,9 @@ import websockets
from solders.pubkey import Pubkey
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from config import (
PUMP_PROGRAM,
)
from config import SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM as ATA_PROGRAM_ID
from config import SYSTEM_TOKEN_PROGRAM as TOKEN_PROGRAM_ID
from config import (
WSS_ENDPOINT,
)
from core.pubkeys import PumpAddresses, SystemAddresses
WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey:
@@ -28,16 +23,16 @@ def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey
derived_address, _ = Pubkey.find_program_address(
[
bytes(bonding_curve),
bytes(TOKEN_PROGRAM_ID),
bytes(SystemAddresses.TOKEN_PROGRAM),
bytes(mint),
],
ATA_PROGRAM_ID,
SystemAddresses.ASSOCIATED_TOKEN_PROGRAM,
)
return derived_address
# Load the IDL JSON file
with open("../idl/pump_fun_idl.json", "r") as f:
with open("idl/pump_fun_idl.json") as f:
idl = json.load(f)
# Extract the "create" instruction definition
@@ -102,13 +97,15 @@ async def listen_for_new_tokens():
"id": 1,
"method": "logsSubscribe",
"params": [
{"mentions": [str(PUMP_PROGRAM)]},
{"mentions": [str(PumpAddresses.PROGRAM)]},
{"commitment": "processed"},
],
}
)
await websocket.send(subscription_message)
print(f"Listening for new token creations from program: {PUMP_PROGRAM}")
print(
f"Listening for new token creations from program: {PumpAddresses.PROGRAM}"
)
# Wait for subscription confirmation
response = await websocket.recv()
@@ -165,7 +162,7 @@ async def listen_for_new_tokens():
)
except Exception as e:
print(f"Failed to decode: {log}")
print(f"Error: {str(e)}")
print(f"Error: {e!s}")
except Exception as e:
print(f"An error occurred while processing message: {e}")
learning-examples/listen_to_raydium_migration.py
+12 -7
View File
@@ -1,14 +1,15 @@
import asyncio
import base64
import json
import os
import sys
import websockets
from solders.pubkey import Pubkey
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from config import PUMP_LIQUIDITY_MIGRATOR, WSS_ENDPOINT
from core.pubkeys import PumpAddresses
WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
def process_initialize2_transaction(data):
@@ -32,7 +33,7 @@ def process_initialize2_transaction(data):
print(f"\nError: Not enough account keys (found {len(account_keys)})")
except Exception as e:
print(f"\nError: {str(e)}")
print(f"\nError: {e!s}")
async def listen_for_events():
@@ -45,7 +46,11 @@ async def listen_for_events():
"id": 1,
"method": "blockSubscribe",
"params": [
{"mentionsAccountOrProgram": str(PUMP_LIQUIDITY_MIGRATOR)},
{
"mentionsAccountOrProgram": str(
PumpAddresses.LIQUIDITY_MIGRATOR
)
},
{
"commitment": "confirmed",
"encoding": "json",
@@ -93,13 +98,13 @@ async def listen_for_events():
process_initialize2_transaction(tx)
break
except asyncio.TimeoutError:
except TimeoutError:
print("\nChecking connection...")
print("Connection alive")
continue
except Exception as e:
print(f"\nConnection error: {str(e)}")
print(f"\nConnection error: {e!s}")
print("Retrying in 5 seconds...")
await asyncio.sleep(5)
learning-examples/manual_buy.py
+4 -4
View File
@@ -12,10 +12,10 @@ from construct import Flag, Int64ul, Struct
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
from solana.rpc.types import TxOpts
from solana.transaction import Message
from solders.compute_budget import set_compute_unit_price
from solders.instruction import AccountMeta, Instruction
from solders.keypair import Keypair
from solders.message import Message
from solders.pubkey import Pubkey
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction, VersionedTransaction
@@ -42,8 +42,8 @@ SOL = Pubkey.from_string("So11111111111111111111111111111111111111112")
LAMPORTS_PER_SOL = 1_000_000_000
# RPC ENDPOINTS
RPC_ENDPOINT = "ENTER_YOUR_CHAINSTACK_HTTP_ENDPOINT"
RPC_WEBSOCKET = "ENTER_YOUR_CHAINSTACK_WS_ENDPOINT"
RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
RPC_WEBSOCKET = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
class BondingCurveState:
@@ -92,7 +92,7 @@ async def buy_token(
slippage: float = 0.25,
max_retries=5,
):
private_key = base58.b58decode("ENTER_PRIVATE_KEY")
private_key = base58.b58decode(os.environ.get("SOLANA_PRIVATE_KEY"))
payer = Keypair.from_bytes(private_key)
async with AsyncClient(RPC_ENDPOINT) as client:
learning-examples/manual_sell.py
+4 -6
View File
@@ -1,19 +1,17 @@
import asyncio
import base64
import json
import os
import struct
import base58
import spl.token.instructions as spl_token
from construct import Flag, Int64ul, Struct
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
from solana.rpc.types import TxOpts
from solana.transaction import Transaction
from solders.instruction import AccountMeta, Instruction
from solders.keypair import Keypair
from solders.pubkey import Pubkey
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction
from spl.token.instructions import get_associated_token_address
# Here and later all the discriminators are precalculated. See learning-examples/discriminator.py
@@ -39,7 +37,7 @@ UNIT_PRICE = 10_000_000
UNIT_BUDGET = 100_000
# RPC endpoint
RPC_ENDPOINT = "SOLANA_NODE_RPC_ENDPOINT"
RPC_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
class BondingCurveState:
@@ -94,7 +92,7 @@ async def sell_token(
slippage: float = 0.25,
max_retries=5,
):
private_key = base58.b58decode("SOLANA_PRIVATE_KEY")
private_key = base58.b58decode(os.environ.get("SOLANA_PRIVATE_KEY"))
payer = Keypair.from_bytes(private_key)
async with AsyncClient(RPC_ENDPOINT) as client:
pyproject.toml
+45 -1
View File
@@ -26,13 +26,57 @@ dev = [
pump_bot = "cli:sync_main"
[tool.ruff]
exclude = [
".bzr",
".direnv",
".eggs",
".git",
".git-rewrite",
".hg",
".ipynb_checkpoints",
".mypy_cache",
".nox",
".pants.d",
".pyenv",
".pytest_cache",
".pytype",
".ruff_cache",
".svn",
".tox",
".venv",
".vscode",
"__pypackages__",
"_build",
"buck-out",
"build",
"dist",
"node_modules",
"site-packages",
"venv",
]
line-length = 88
indent-width = 4
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "N", "B", "A", "C4", "T10", "ARG", "PTH"]
select = [
"E", "F", "I", "UP", "N", "B", "A", "C4", "T10", "ARG", "PTH",
"ANN", # type annotations
"S", # security best practices
"BLE", # blind except statements
"FBT", # boolean trap parameters
"C90", # complexity metrics
"TRY", # exception handling best practices
"SLF", # private member access
"TCH", # type checking issues
"RUF", # Ruff-specific rules
"ERA", # eradicate commented-out code
"PL", # pylint conventions
]
ignore = ["E501"]
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
line-ending = "auto"