pumpfun_api_handler.py

"""
Pump.fun API Handler Module
Handles all Pump.fun blockchain interactions
"""
import os
import struct
from typing import Dict, Any, Optional

from solders.pubkey import Pubkey
from solders.keypair import Keypair
from solders.system_program import ID as SYS_PROGRAM_ID
from solana.rpc.async_api import AsyncClient
from solders.transaction import Transaction
from solders.instruction import Instruction as TransactionInstruction
from solders.instruction import AccountMeta
from solana.rpc.types import TxOpts
from solders.message import Message
from solders.system_program import CreateAccountParams, create_account
from spl.token.constants import TOKEN_PROGRAM_ID, ASSOCIATED_TOKEN_PROGRAM_ID
from spl.token.instructions import get_associated_token_address


LAMPORTS_PER_SOL = 1_000_000_000

# Pump.fun Program Constants
PUMPFUN_PROGRAM_ID = Pubkey.from_string(
    "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
)

GLOBAL_PDA, GLOBAL_BUMP = Pubkey.find_program_address(
    [b"global"], PUMPFUN_PROGRAM_ID
)

BUY_DISCRIMINATOR = bytes([0x66, 0x06, 0x3d, 0x12, 0x01, 0xda, 0xeb, 0xea])
CREATE_DISCRIMINATOR = bytes([0x18, 0x1e, 0xc8, 0x28, 0x05, 0x1c, 0x07, 0x77])
SELL_DISCRIMINATOR = bytes([0x33, 0xe6, 0x85, 0xa4, 0x01, 0x7f, 0x83, 0xad])

EVENT_AUTHORITY = Pubkey.from_string(
    "Ce6TQqeHC9p8KetsN6JsjHK7UTZk7nasjjnr7XxXp9F1"
)

MPL_TOKEN_METADATA = Pubkey.from_string(
    "metaqbxxUerdq28cj1RbAWkYQm3ybzjb6a8bt518x1s"
)

RENT_SYSVAR = Pubkey.from_string(
    "SysvarRent111111111111111111111111111111111"
)

MINT_AUTHORITY, _ = Pubkey.find_program_address(
    [b"mint-authority"], PUMPFUN_PROGRAM_ID
)


class PumpFunHandler:
    """Handles Pump.fun token creation and trading"""
    
    def __init__(self, private_key: Optional[str] = None, rpc_url: Optional[str] = None, testing: bool = False):
        self.private_key = private_key or os.getenv("SOLANA_PRIVATE_KEY")
        self.rpc_url = rpc_url or os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com")
        self.testing = testing
        
        if not self.private_key:
            raise ValueError("Solana private key required")
        
        self.wallet = Keypair.from_base58_string(self.private_key)
    
    async def _derive_bonding_curve(self, mint: Pubkey) -> Pubkey:
        pda, _ = Pubkey.find_program_address(
            [b"bonding-curve", bytes(mint)],
            PUMPFUN_PROGRAM_ID
        )
        return pda
    
    async def _derive_associated_bonding_curve(self, bonding_curve: Pubkey, mint: Pubkey) -> Pubkey:
        pda, _ = Pubkey.find_program_address(
            [
                bytes(bonding_curve),
                bytes(TOKEN_PROGRAM_ID),
                bytes(mint),
            ],
            ASSOCIATED_TOKEN_PROGRAM_ID
        )
        return pda
    
    async def _derive_metadata(self, mint: Pubkey) -> Pubkey:
        pda, _ = Pubkey.find_program_address(
            [
                b"metadata",
                bytes(MPL_TOKEN_METADATA),
                bytes(mint),
            ],
            MPL_TOKEN_METADATA
        )
        return pda
    
    async def create_token(
        self,
        token_params: Dict[str, Any],
        metadata_uri: str
    ) -> Dict[str, Any]:
        """Create a new token on Pump.fun"""
        name = token_params["name"]
        symbol = token_params["symbol"]
        
        if not metadata_uri.startswith("ipfs://"):
            raise ValueError(
                "metadata_uri must start with 'ipfs://'. "
                "Upload metadata via pump.fun in a browser and paste the URI."
            )
        
        name_b = name.encode()
        symbol_b = symbol.encode()
        uri_b = metadata_uri.encode()
        
        metadata_bytes = (
            struct.pack("<I", len(name_b)) + name_b +
            struct.pack("<I", len(symbol_b)) + symbol_b +
            struct.pack("<I", len(uri_b)) + uri_b
        )
        
        mint_kp = Keypair()
        mint = mint_kp.pubkey()
        
        bonding_curve = await self._derive_bonding_curve(mint)
        associated_bonding_curve = await self._derive_associated_bonding_curve(
            bonding_curve, mint
        )
        metadata = await self._derive_metadata(mint)
        user = self.wallet.pubkey()
        
        accounts = [
            AccountMeta(pubkey=mint, is_signer=True, is_writable=True),
            AccountMeta(pubkey=MINT_AUTHORITY, is_signer=False, is_writable=False),
            AccountMeta(pubkey=bonding_curve, is_signer=False, is_writable=True),
            AccountMeta(pubkey=associated_bonding_curve, is_signer=False, is_writable=True),
            AccountMeta(pubkey=GLOBAL_PDA, is_signer=False, is_writable=False),
            AccountMeta(pubkey=MPL_TOKEN_METADATA, is_signer=False, is_writable=False),
            AccountMeta(pubkey=metadata, is_signer=False, is_writable=True),
            AccountMeta(pubkey=user, is_signer=True, is_writable=True),
            AccountMeta(pubkey=SYS_PROGRAM_ID, is_signer=False, is_writable=False),
            AccountMeta(pubkey=TOKEN_PROGRAM_ID, is_signer=False, is_writable=False),
            AccountMeta(pubkey=ASSOCIATED_TOKEN_PROGRAM_ID, is_signer=False, is_writable=False),
            AccountMeta(pubkey=RENT_SYSVAR, is_signer=False, is_writable=False),
            AccountMeta(pubkey=EVENT_AUTHORITY, is_signer=False, is_writable=False),
            AccountMeta(pubkey=PUMPFUN_PROGRAM_ID, is_signer=False, is_writable=False),
        ]
        
        instruction_data = CREATE_DISCRIMINATOR + metadata_bytes
        
        ix = TransactionInstruction(
            program_id=PUMPFUN_PROGRAM_ID,
            accounts=accounts,
            data=instruction_data
        )
        
        client = AsyncClient(self.rpc_url)
        recent_blockhash = (await client.get_latest_blockhash()).value.blockhash
        
        msg = Message.new_with_blockhash(
            [ix],
            self.wallet.pubkey(),
            recent_blockhash
        )
        
        tx = Transaction([self.wallet, mint_kp], msg)
        
        opts = TxOpts(skip_preflight=False, preflight_commitment="confirmed")
        result = await client.send_transaction(tx, opts=opts)
        signature = str(result.value)
        
        await client.confirm_transaction(signature, commitment="confirmed")
        await client.close()
        
        return {
            "mint": str(mint),
            "signature": signature,
            "name": name,
            "symbol": symbol,
            "metadata_uri": metadata_uri
        }
    
    async def get_token_data(self, mint: str) -> Dict[str, Any]:
        """Fetch current token data (mock implementation)"""
        # In production, this would fetch real data from Pump.fun API
        return {
            "price": 0.0001,
            "marketCap": 10000,
            "bondingCurveProgress": 0.15,
            "holders": 50,
            "volume24h": 500,
            "treasuryBalanceSol": 2.5
        }
    
    async def execute_buyback(self, mint: str, amount_sol: float):
        """Execute a token buyback (mock implementation)"""
        print(f"💰 BUYBACK: {amount_sol} SOL for {mint}")
        # Real implementation would execute buy transaction
    
    async def execute_sell(self, mint: str, amount_sol: float):
        """Execute a token sell (mock implementation)"""
        print(f"💸 SELL: {amount_sol} SOL worth of {mint}")
        # Real implementation would execute sell transaction