diff --git a/learning-examples/blockSubscribe_extract_transactions.py b/learning-examples/blockSubscribe_extract_transactions.py index 474f94b..671fb49 100644 --- a/learning-examples/blockSubscribe_extract_transactions.py +++ b/learning-examples/blockSubscribe_extract_transactions.py @@ -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__": diff --git a/learning-examples/check_boding_curve_status.py b/learning-examples/check_boding_curve_status.py index 86a97ca..dfc1cb3 100644 --- a/learning-examples/check_boding_curve_status.py +++ b/learning-examples/check_boding_curve_status.py @@ -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(" 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:") diff --git a/learning-examples/compute_associated_bonding_curve.py b/learning-examples/compute_associated_bonding_curve.py index d354eb1..12343ec 100644 --- a/learning-examples/compute_associated_bonding_curve.py +++ b/learning-examples/compute_associated_bonding_curve.py @@ -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( diff --git a/learning-examples/decode_from_blockSubscribe.py b/learning-examples/decode_from_blockSubscribe.py index 4124568..1bebc96 100644 --- a/learning-examples/decode_from_blockSubscribe.py +++ b/learning-examples/decode_from_blockSubscribe.py @@ -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 ") - 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) diff --git a/learning-examples/decode_from_getAccountInfo.py b/learning-examples/decode_from_getAccountInfo.py index 1a62ac3..c342607 100644 --- a/learning-examples/decode_from_getAccountInfo.py +++ b/learning-examples/decode_from_getAccountInfo.py @@ -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 diff --git a/learning-examples/decode_from_getTransaction.py b/learning-examples/decode_from_getTransaction.py index 5be9b6d..630d8f3 100644 --- a/learning-examples/decode_from_getTransaction.py +++ b/learning-examples/decode_from_getTransaction.py @@ -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 ") - 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 diff --git a/learning-examples/fetch_price.py b/learning-examples/fetch_price.py index e2eb460..ac8a1b9 100644 --- a/learning-examples/fetch_price.py +++ b/learning-examples/fetch_price.py @@ -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(" 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}") diff --git a/learning-examples/listen_to_raydium_migration.py b/learning-examples/listen_to_raydium_migration.py index b57493e..4d3e823 100644 --- a/learning-examples/listen_to_raydium_migration.py +++ b/learning-examples/listen_to_raydium_migration.py @@ -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) diff --git a/learning-examples/manual_buy.py b/learning-examples/manual_buy.py index 91d8616..cb00e29 100644 --- a/learning-examples/manual_buy.py +++ b/learning-examples/manual_buy.py @@ -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: diff --git a/learning-examples/manual_sell.py b/learning-examples/manual_sell.py index e63eb51..ab4aa61 100644 --- a/learning-examples/manual_sell.py +++ b/learning-examples/manual_sell.py @@ -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: diff --git a/pyproject.toml b/pyproject.toml index 951a7bd..e28e445 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" \ No newline at end of file