### Install pumpswapamm Source: https://context7.com/flock4h/pumpswapamm/llms.txt Install the pumpswapamm library using pip. Ensure you have the required dependencies like solana-py and solders. ```bash pip install pumpswapamm ``` -------------------------------- ### Execute Complete Buy and Sell Workflow Source: https://context7.com/flock4h/pumpswapamm/llms.txt Demonstrates a full trading cycle including pool state fetching, price calculation, and executing buy/sell transactions. Ensure pool data is refreshed before selling to account for balance changes. ```python import asyncio from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Processed from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap, fetch_pool_state async def complete_trading_workflow(): # Configuration RPC_ENDPOINT = "https://api.mainnet-beta.solana.com" PRIVATE_KEY = "your_private_key_here" POOL_ADDRESS = "2bEXGX6B3gc23B6sFmESaPgC5wyCV1hzCFxpDtJk7DE1" MINT_ADDRESS = "L2aKgxwNSGn6i5njrDqtFxan82uidsfjXHrYNSspump" # Initialize async_client = AsyncClient(RPC_ENDPOINT) keypair = Keypair.from_base58_string(PRIVATE_KEY) client = PumpSwap(async_client) try: # Step 1: Fetch pool state pool_keys, pool_type = await fetch_pool_state(POOL_ADDRESS, async_client) if not pool_keys: print("Failed to fetch pool state") return # Step 2: Get current price and reserves base_price, base_balance, quote_balance = await client.fetch_pool_base_price(POOL_ADDRESS) print(f"Current price: {base_price} SOL/token") # Step 3: Get token decimals mint_info = await async_client.get_account_info_json_parsed( Pubkey.from_string(MINT_ADDRESS), commitment=Processed ) decimals = mint_info.value.data.parsed['info']['decimals'] # Step 4: Prepare pool data pool_data = { "pool_pubkey": Pubkey.from_string(POOL_ADDRESS), "token_base": Pubkey.from_string(pool_keys["base_mint"]), "token_quote": Pubkey.from_string(pool_keys["quote_mint"]), "pool_base_token_account": pool_keys["pool_base_token_account"], "pool_quote_token_account": pool_keys["pool_quote_token_account"], "base_balance_tokens": base_balance, "quote_balance_sol": quote_balance, "decimals_base": decimals, } if pool_type == "NEW": pool_data["coin_creator"] = Pubkey.from_string(pool_keys["coin_creator"]) # Step 5: Execute buy print("\n--- Executing Buy ---") buy_result = await client.buy( pool_data=pool_data, sol_amount=0.002, keypair=keypair, pool_type=pool_type, slippage_pct=10, fee_sol=0.0005, debug_prints=True, ) confirmed, tx_sig, _, tokens = buy_result print(f"Buy confirmed: {confirmed}, TX: {tx_sig}") if not confirmed: print("Buy failed, exiting") return # Step 6: Wait before selling print("\nWaiting 10 seconds before selling...") await asyncio.sleep(10) # Step 7: Refresh pool data for accurate sell base_price, base_balance, quote_balance = await client.fetch_pool_base_price(POOL_ADDRESS) pool_data["base_balance_tokens"] = base_balance pool_data["quote_balance_sol"] = quote_balance # Step 8: Execute sell print("\n--- Executing Sell ---") sell_result = await client.sell( pool_data=pool_data, sell_pct=100, keypair=keypair, pool_type=pool_type, slippage_pct=10, fee_sol=0.0005, debug_prints=True, ) confirmed, tx_sig, _, sol_out = sell_result print(f"Sell confirmed: {confirmed}, TX: {tx_sig}") finally: await client.close() asyncio.run(complete_trading_workflow()) ``` -------------------------------- ### Initialize PumpSwap Class Source: https://context7.com/flock4h/pumpswapamm/llms.txt Initialize the PumpSwap class with an async Solana RPC client and a keypair. This is the main entry point for interacting with AMM pools. ```python import asyncio from solana.rpc.async_api import AsyncClient from solders.keypair import Keypair from pumpswapamm import PumpSwap, fetch_pool_state # Initialize RPC client and keypair RPC_ENDPOINT = "https://api.mainnet-beta.solana.com" PRIVATE_KEY = "your_base58_private_key_here" async_client = AsyncClient(RPC_ENDPOINT) keypair = Keypair.from_base58_string(PRIVATE_KEY) # Create PumpSwap instance client = PumpSwap(async_client) # Clean up when done await client.close() ``` -------------------------------- ### Execute Buy Transaction Source: https://context7.com/flock4h/pumpswapamm/llms.txt Performs a swap of SOL for tokens. Requires fetching pool state and token decimals before execution. ```python import asyncio from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Processed from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap, fetch_pool_state async def buy_tokens(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) pool_address = "2bEXGX6B3gc23B6sFmESaPgC5wyCV1hzCFxpDtJk7DE1" mint_address = "L2aKgxwNSGn6i5njrDqtFxan82uidsfjXHrYNSspump" # Fetch pool state and price pool_keys, pool_type = await fetch_pool_state(pool_address, async_client) base_price, base_balance_tokens, quote_balance_sol = await client.fetch_pool_base_price(pool_address) # Get token decimals mint_info = await async_client.get_account_info_json_parsed( Pubkey.from_string(mint_address), commitment=Processed ) decimals = mint_info.value.data.parsed['info']['decimals'] # Prepare pool data dict pool_data = { "pool_pubkey": Pubkey.from_string(pool_address), "token_base": Pubkey.from_string(pool_keys["base_mint"]), "token_quote": Pubkey.from_string(pool_keys["quote_mint"]), "pool_base_token_account": pool_keys["pool_base_token_account"], "pool_quote_token_account": pool_keys["pool_quote_token_account"], "base_balance_tokens": base_balance_tokens, "quote_balance_sol": quote_balance_sol, "decimals_base": decimals, "coin_creator": Pubkey.from_string(pool_keys["coin_creator"]), # Required for NEW pools } # Execute buy: spend 0.002 SOL to buy tokens confirmed, tx_sig, pool_type, tokens_bought = await client.buy( pool_data=pool_data, sol_amount=0.002, # Amount of SOL to spend keypair=keypair, pool_type=pool_type, # "NEW" or "OLD" slippage_pct=10, # 10% slippage tolerance fee_sol=0.0005, # Priority fee in SOL debug_prints=True, # Print transaction URL ) if confirmed: print(f"Buy successful! TX: https://solscan.io/tx/{tx_sig}") print(f"Tokens received: {tokens_bought / (10 ** decimals)}") else: print("Transaction failed") await client.close() asyncio.run(buy_tokens()) ``` -------------------------------- ### Create Associated Token Account with PumpSwap Source: https://context7.com/flock4h/pumpswapamm/llms.txt Checks for an existing ATA and returns a creation instruction if necessary. Requires a valid AsyncClient and Keypair. ```python import asyncio from solana.rpc.async_api import AsyncClient from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap async def ensure_ata_exists(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) token_mint = Pubkey.from_string("token_mint_address_here") # Returns instruction if ATA needs to be created, None otherwise create_ata_ix = await client.create_ata_if_needed( owner=keypair.pubkey(), mint=token_mint, ) if create_ata_ix: print("ATA needs to be created - include this instruction in your transaction") else: print("ATA already exists") await client.close() asyncio.run(ensure_ata_exists()) ``` -------------------------------- ### Create PumpSwap Liquidity Pool Source: https://context7.com/flock4h/pumpswapamm/llms.txt Creates a new PumpSwap liquidity pool. The caller becomes the creator and initial LP token holder. Requires initial liquidity amounts for both base and quote tokens. ```python import asyncio from solana.rpc.async_api import AsyncClient from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap async def create_new_pool(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) mint_address = "8oubm4nEgTFFa6SQWUoav9hpGt6MCrWQt5yXUBWEpump" # Create pool with initial liquidity pool_pda = await client.create_pool( base_mint=Pubkey.from_string(mint_address), base_amount_tokens=990_000_000, # 990M tokens (UI amount, not raw) quote_amount_sol=0.01, # 0.01 SOL initial liquidity keypair=keypair, decimals_base=6, # Token decimals index=0, # Pool index (usually 0) fee_sol=0.001, # Priority fee debug_prints=True, ) if pool_pda: print(f"Pool created successfully!") print(f"Pool PDA: {pool_pda}") else: print("Pool creation failed") await client.close() asyncio.run(create_new_pool()) ``` -------------------------------- ### create_pool Source: https://context7.com/flock4h/pumpswapamm/llms.txt Creates a new PumpSwap liquidity pool with initial liquidity. ```APIDOC ## create_pool ### Description Creates a brand-new PumpSwap liquidity pool. The calling wallet becomes the pool creator and initial LP token holder. ### Parameters - **base_mint** (Pubkey) - Required - The mint address of the token. - **base_amount_tokens** (int) - Required - Initial amount of tokens to deposit. - **quote_amount_sol** (float) - Required - Initial amount of SOL to deposit. - **keypair** (Keypair) - Required - The wallet keypair creating the pool. - **decimals_base** (int) - Required - Token decimals. - **index** (int) - Required - Pool index. - **fee_sol** (float) - Optional - Priority fee. - **debug_prints** (bool) - Optional - Enable debug logs. ### Response - **pool_pda** (Pubkey) - The derived address of the created pool. ``` -------------------------------- ### Execute Sell Transaction Source: https://context7.com/flock4h/pumpswapamm/llms.txt Performs a swap of tokens for SOL. Allows specifying a percentage of holdings to sell. ```python import asyncio from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Processed from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap, fetch_pool_state async def sell_tokens(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) pool_address = "2bEXGX6B3gc23B6sFmESaPgC5wyCV1hzCFxpDtJk7DE1" mint_address = "L2aKgxwNSGn6i5njrDqtFxan82uidsfjXHrYNSspump" # Fetch pool state and reserves pool_keys, pool_type = await fetch_pool_state(pool_address, async_client) base_price, base_balance_tokens, quote_balance_sol = await client.fetch_pool_base_price(pool_address) # Get token decimals mint_info = await async_client.get_account_info_json_parsed( Pubkey.from_string(mint_address), commitment=Processed ) decimals = mint_info.value.data.parsed['info']['decimals'] pool_data = { "pool_pubkey": Pubkey.from_string(pool_address), "token_base": Pubkey.from_string(pool_keys["base_mint"]), "token_quote": Pubkey.from_string(pool_keys["quote_mint"]), "pool_base_token_account": pool_keys["pool_base_token_account"], "pool_quote_token_account": pool_keys["pool_quote_token_account"], "base_balance_tokens": base_balance_tokens, "quote_balance_sol": quote_balance_sol, "decimals_base": decimals, "coin_creator": Pubkey.from_string(pool_keys["coin_creator"]), } # Sell 100% of token holdings confirmed, tx_sig, pool_type, sol_received = await client.sell( pool_data=pool_data, sell_pct=100, # Percentage of holdings to sell (1-100) keypair=keypair, pool_type=pool_type, slippage_pct=10, # 10% slippage tolerance fee_sol=0.0005, # Priority fee debug_prints=True, ) if confirmed: print(f"Sell successful! TX: https://solscan.io/tx/{tx_sig}") print(f"SOL received: {sol_received}") else: print("Transaction failed or no tokens to sell") await client.close() asyncio.run(sell_tokens()) ``` -------------------------------- ### Handle Reversed Pools (WSOL as Base) in PumpSwap Source: https://context7.com/flock4h/pumpswapamm/llms.txt Use `reversed_buy` to buy WSOL by selling another token, or `reversed_sell` to sell WSOL to buy another token. These methods are for pools where WSOL is the base mint. ```python import asyncio from solana.rpc.async_api import AsyncClient from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap, fetch_pool_state, WSOL_MINT async def handle_reversed_pool(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) # For reversed pools (WSOL is base, TOKEN is quote) pool_address = "reversed_pool_address_here" pool_keys, pool_type = await fetch_pool_state(pool_address, async_client) base_price, base_balance_tokens, quote_balance_sol = await client.fetch_pool_base_price(pool_address) pool_data = { "pool_pubkey": Pubkey.from_string(pool_address), "token_base": Pubkey.from_string(pool_keys["base_mint"]), # WSOL "token_quote": Pubkey.from_string(pool_keys["quote_mint"]), # TOKEN "pool_base_token_account": pool_keys["pool_base_token_account"], "pool_quote_token_account": pool_keys["pool_quote_token_account"], "base_balance_tokens": base_balance_tokens, "quote_balance_sol": quote_balance_sol, "decimals_base": 9, # WSOL decimals "decimals_quote": 6, # Token decimals "coin_creator": Pubkey.from_string(pool_keys["coin_creator"]), } # reversed_buy: Buy WSOL by selling TOKEN confirmed, tx_sig, pool_type, wsol_bought = await client.reversed_buy( pool_data=pool_data, sell_pct=50, # Sell 50% of token balance keypair=keypair, pool_type=pool_type, slippage_pct=10, fee_sol=0.0001, debug_prints=True, ) # reversed_sell: Sell WSOL to buy TOKEN confirmed, tx_sig, pool_type, tokens_bought = await client.reversed_sell( pool_data=pool_data, sol_amount=0.01, # Amount of SOL/WSOL to spend keypair=keypair, pool_type=pool_type, slippage_pct=10, fee_sol=0.0001, debug_prints=True, ) await client.close() asyncio.run(handle_reversed_pool()) ``` -------------------------------- ### Fetch Pool Base Price Source: https://context7.com/flock4h/pumpswapamm/llms.txt Fetches the current base price of a pool by querying reserve balances. Returns the price in quote tokens per base token along with the current reserve amounts. ```python import asyncio from solana.rpc.async_api import AsyncClient from pumpswapamm import PumpSwap async def get_price(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") client = PumpSwap(async_client) pool_address = "2bEXGX6B3gc23B6sFmESaPgC5wyCV1hzCFxpDtJk7DE1" # Returns (price, base_balance, quote_balance) base_price, base_balance_tokens, quote_balance_sol = await client.fetch_pool_base_price(pool_address) print(f"Price: {base_price} SOL per token") print(f"Base Reserve: {base_balance_tokens} tokens") print(f"Quote Reserve: {quote_balance_sol} SOL") await client.close() # Expected output: # Price: 0.000001234 SOL per token # Base Reserve: 1000000000.0 tokens # Quote Reserve: 1234.5 SOL asyncio.run(get_price()) ``` -------------------------------- ### Fetch PumpSwap Pool State Source: https://context7.com/flock4h/pumpswapamm/llms.txt Fetches and parses the on-chain state of a PumpSwap pool using its address and an async Solana RPC client. Supports both old and new pool formats. ```python import asyncio from solana.rpc.async_api import AsyncClient from solders.pubkey import Pubkey from pumpswapamm import fetch_pool_state, convert_pool_keys async def get_pool_info(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") pool_address = "2bEXGX6B3gc23B6sFmESaPgC5wyCV1hzCFxpDtJk7DE1" # Fetch pool state - returns tuple of (pool_keys, pool_type) pool_keys, pool_type = await fetch_pool_state(pool_address, async_client) if pool_keys: print(f"Pool Type: {pool_type}") # "NEW" or "OLD" print(f"Base Mint: {pool_keys['base_mint']}") print(f"Quote Mint: {pool_keys['quote_mint']}") print(f"LP Mint: {pool_keys['lp_mint']}") print(f"Pool Base Token Account: {pool_keys['pool_base_token_account']}") print(f"Pool Quote Token Account: {pool_keys['pool_quote_token_account']}") print(f"LP Supply: {pool_keys['lp_supply']}") if pool_type == "NEW": print(f"Coin Creator: {pool_keys['coin_creator']}") await async_client.close() # Expected output: # Pool Type: NEW # Base Mint: L2aKgxwNSGn6i5njrDqtFxan82uidsfjXHrYNSspump # Quote Mint: So11111111111111111111111111111111111111112 # LP Mint: ... # ... asyncio.run(get_pool_info()) ``` -------------------------------- ### deposit Source: https://context7.com/flock4h/pumpswapamm/llms.txt Deposits additional liquidity into an existing PumpSwap pool. ```APIDOC ## deposit ### Description Deposits additional liquidity into an existing PumpSwap pool. Automatically calculates the proportional SOL amount needed based on current reserves. ### Parameters - **pool_data** (dict) - Required - Dictionary containing pool state and account information. - **base_amount_tokens** (int) - Required - Amount of tokens to deposit. - **keypair** (Keypair) - Required - The wallet keypair performing the deposit. - **slippage_pct** (float) - Optional - Allowed slippage percentage. - **fee_sol** (float) - Optional - Priority fee. - **sol_cap** (float) - Optional - Maximum SOL to spend. - **debug_prints** (bool) - Optional - Enable debug logs. ### Response - **success** (bool) - Returns true if the deposit was successful. ``` -------------------------------- ### Deposit Liquidity into PumpSwap Pool Source: https://context7.com/flock4h/pumpswapamm/llms.txt Deposits additional liquidity into an existing PumpSwap pool. The required SOL amount is automatically calculated based on current reserves and the provided token amount. Includes safety features like slippage control and a SOL cap. ```python import asyncio from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Processed from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap, fetch_pool_state from pumpswapamm.fetch_reserves import fetch_pool_base_price async def deposit_liquidity(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) pool_address = "your_pool_address_here" mint_address = "your_token_mint_here" # Fetch pool data pool_keys, pool_type = await fetch_pool_state(pool_address, async_client) _, base_bal, quote_bal = await fetch_pool_base_price(pool_keys, async_client) # Get token decimals mint_info = await async_client.get_account_info_json_parsed( Pubkey.from_string(mint_address), commitment=Processed ) decimals = mint_info.value.data.parsed['info']['decimals'] pool_data = { "pool_pubkey": Pubkey.from_string(pool_address), "token_base": Pubkey.from_string(pool_keys["base_mint"]), "token_quote": Pubkey.from_string(pool_keys["quote_mint"]), "lp_mint": pool_keys["lp_mint"], "pool_base_token_account": pool_keys["pool_base_token_account"], "pool_quote_token_account": pool_keys["pool_quote_token_account"], "base_balance_tokens": base_bal, "quote_balance_sol": quote_bal, "decimals_base": decimals, "coin_creator": Pubkey.from_string(pool_keys["coin_creator"]), } # Deposit 2M tokens (SOL amount calculated automatically) success = await client.deposit( pool_data=pool_data, base_amount_tokens=2_000_000, # Tokens to deposit keypair=keypair, slippage_pct=1.0, # 1% slippage fee_sol=0.0002, # Priority fee sol_cap=0.1, # Max SOL to spend (safety cap) debug_prints=True, ) print(f"Deposit {'successful' if success else 'failed'}") await client.close() asyncio.run(deposit_liquidity()) ``` -------------------------------- ### derive_pool_address Source: https://context7.com/flock4h/pumpswapamm/llms.txt Statically derives the PDA for a pool given its parameters. ```APIDOC ## derive_pool_address ### Description Statically derives the PDA (Program Derived Address) for a pool given its parameters. Useful when you know the pool parameters but don't have the address. ### Parameters - **creator** (Pubkey) - Required - The creator's public key. - **base_mint** (Pubkey) - Required - The base token mint address. - **quote_mint** (Pubkey) - Required - The quote token mint address. - **index** (int) - Required - Pool index. ### Response - **pool_pda** (Pubkey) - The derived pool address. ``` -------------------------------- ### Withdraw Liquidity from PumpSwap Pool Source: https://context7.com/flock4h/pumpswapamm/llms.txt Use this function to burn LP tokens and withdraw proportional amounts of base and quote tokens from a PumpSwap pool. Ensure you have your private key and the correct pool address. ```python import asyncio from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Processed from solders.keypair import Keypair from solders.pubkey import Pubkey from pumpswapamm import PumpSwap, fetch_pool_state from pumpswapamm.fetch_reserves import fetch_pool_base_price async def withdraw_liquidity(): async_client = AsyncClient("https://api.mainnet-beta.solana.com") keypair = Keypair.from_base58_string("your_private_key_here") client = PumpSwap(async_client) pool_address = "your_pool_address_here" mint_address = "your_token_mint_here" # Fetch pool data pool_keys, pool_type = await fetch_pool_state(pool_address, async_client) _, base_bal, quote_bal = await fetch_pool_base_price(pool_keys, async_client) mint_info = await async_client.get_account_info_json_parsed( Pubkey.from_string(mint_address), commitment=Processed ) decimals = mint_info.value.data.parsed['info']['decimals'] pool_data = { "pool_pubkey": Pubkey.from_string(pool_address), "token_base": Pubkey.from_string(pool_keys["base_mint"]), "token_quote": Pubkey.from_string(pool_keys["quote_mint"]), "lp_mint": pool_keys["lp_mint"], "pool_base_token_account": pool_keys["pool_base_token_account"], "pool_quote_token_account": pool_keys["pool_quote_token_account"], "base_balance_tokens": base_bal, "quote_balance_sol": quote_bal, "decimals_base": decimals, "coin_creator": Pubkey.from_string(pool_keys["coin_creator"]), } # Withdraw 100% of LP position success = await client.withdraw( pool_data=pool_data, withdraw_pct=100, # Percentage of LP tokens to burn fee_sol=0.0002, # Priority fee debug_prints=True, keypair=keypair, ) print(f"Withdrawal {'successful' if success else 'failed'}") await client.close() asyncio.run(withdraw_liquidity()) ``` -------------------------------- ### Derive PumpSwap Pool Address Source: https://context7.com/flock4h/pumpswapamm/llms.txt Statically derives the Program Derived Address (PDA) for a PumpSwap pool using its parameters. This is useful when the pool address is needed without querying the blockchain. ```python from solders.pubkey import Pubkey from solders.keypair import Keypair from pumpswapamm import PumpSwap, WSOL_MINT # Derive pool address without querying the chain keypair = Keypair.from_base58_string("your_private_key_here") mint_address = "8oubm4nEgTFFa6SQWUoav9hpGt6MCrWQt5yXUBWEpump" pool_pda = PumpSwap.derive_pool_address( creator=keypair.pubkey(), base_mint=Pubkey.from_string(mint_address), quote_mint=WSOL_MINT, index=0, # Pool index ) print(f"Derived Pool Address: {pool_pda}") ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.