diff --git a/buy.py b/.archive/buy.py similarity index 99% rename from buy.py rename to .archive/buy.py index 4e35ee1..7b78290 100644 --- a/buy.py +++ b/.archive/buy.py @@ -21,6 +21,7 @@ from solders.transaction import VersionedTransaction from spl.token.instructions import get_associated_token_address from config import * +from trade import trade # Here and later all the discriminators are precalculated. See learning-examples/discriminator.py EXPECTED_DISCRIMINATOR = struct.pack(" argparse.Namespace: + """Parse command line arguments. + + Returns: + Parsed arguments + """ + parser = argparse.ArgumentParser(description="Trade tokens on pump.fun.") + parser.add_argument( + "--yolo", action="store_true", help="Run in YOLO mode (continuous trading)" + ) + parser.add_argument( + "--match", + type=str, + help="Only trade tokens with names or symbols matching this string", + ) + parser.add_argument( + "--bro", type=str, help="Only trade tokens created by this user address" + ) + parser.add_argument( + "--marry", action="store_true", help="Only buy tokens, skip selling" + ) + parser.add_argument( + "--amount", + type=float, + help=f"Amount of SOL to spend on each buy (default: {config.BUY_AMOUNT})", + ) + parser.add_argument( + "--buy-slippage", + type=float, + help=f"Buy slippage tolerance (default: {config.BUY_SLIPPAGE})", + ) + parser.add_argument( + "--sell-slippage", + type=float, + help=f"Sell slippage tolerance (default: {config.SELL_SLIPPAGE})", + ) + parser.add_argument("--rpc", type=str, help="Solana RPC endpoint") + parser.add_argument("--wss", type=str, help="Solana WebSocket endpoint") + + parser.add_argument("--wss", type=str, help="Solana WebSocket endpoint") + parser.add_argument( + "--key", + type=str, + help="Solana private key (better to use environment variable)", + ) + parser.add_argument("--log", type=str, help="Log file path") + return parser.parse_args() + + +async def main() -> None: + """Main entry point for the CLI.""" + setup_file_logging("pump_trading.log") + + args = parse_args() + + # Get configuration values, preferring command line args over config.py + rpc_endpoint: str | None = args.rpc or os.environ.get("SOLANA_NODE_RPC_ENDPOINT") + wss_endpoint: str | None = args.wss or os.environ.get("SOLANA_NODE_WSS_ENDPOINT") + private_key: str | None = args.key or os.environ.get("SOLANA_PRIVATE_KEY") + + if not rpc_endpoint or rpc_endpoint.startswith(("http://", "https://")): + logger.error("Invalid RPC endpoint. Must start with http:// or https://") + sys.exit(1) + + if not wss_endpoint or wss_endpoint.startswith(("ws://", "wss://")): + logger.error("Invalid WebSocket endpoint. Must start with ws:// or wss://") + sys.exit(1) + + if not private_key or len(private_key) < 80: + logger.error("Invalid private key. Key appears to be missing or too short") + sys.exit(1) + + buy_amount: float = args.amount if args.amount is not None else config.BUY_AMOUNT + buy_slippage: float = ( + args.buy_slippage if args.buy_slippage is not None else config.BUY_SLIPPAGE + ) + sell_slippage: float = ( + args.sell_slippage if args.sell_slippage is not None else config.SELL_SLIPPAGE + ) + + trader: PumpTrader = PumpTrader( + rpc_endpoint=rpc_endpoint, # type: ignore + wss_endpoint=wss_endpoint, # type: ignore + private_key=private_key, + buy_amount=buy_amount, + buy_slippage=buy_slippage, + sell_slippage=sell_slippage, + max_retries=config.MAX_RETRIES, + ) + + try: + await trader.start( + match_string=args.match, + bro_address=args.bro, + marry_mode=args.marry, + yolo_mode=args.yolo, + ) + except KeyboardInterrupt: + logger.info("Trading stopped by user") + except Exception as e: + logger.error(f"Trading stopped due to error: {str(e)}") + finally: + try: + await trader.solana_client.close() + except Exception: + pass + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/config.py b/config.py index 5d0c72c..677a551 100644 --- a/config.py +++ b/config.py @@ -1,3 +1,9 @@ +""" +Configuration for the pump.fun trading bot. +""" + +from typing import Final + import os from dotenv import load_dotenv @@ -6,23 +12,33 @@ from solders.pubkey import Pubkey load_dotenv() # System & pump.fun addresses -PUMP_PROGRAM = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P") -PUMP_GLOBAL = Pubkey.from_string("4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf") -PUMP_EVENT_AUTHORITY = Pubkey.from_string( +PUMP_PROGRAM: Final[Pubkey] = Pubkey.from_string( + "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P" +) +PUMP_GLOBAL: Final[Pubkey] = Pubkey.from_string( + "4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf" +) +PUMP_EVENT_AUTHORITY: Final[Pubkey] = Pubkey.from_string( "Ce6TQqeHC9p8KetsN6JsjHK7UTZk7nasjjnr7XxXp9F1" ) -PUMP_FEE = Pubkey.from_string("CebN5WGQ4jvEPvsVU4EoHEpgzq1VV7AbicfhtW4xC9iM") -PUMP_LIQUIDITY_MIGRATOR = Pubkey.from_string( +PUMP_FEE: Final[Pubkey] = Pubkey.from_string( + "CebN5WGQ4jvEPvsVU4EoHEpgzq1VV7AbicfhtW4xC9iM" +) +PUMP_LIQUIDITY_MIGRATOR: Final[Pubkey] = Pubkey.from_string( "39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg" ) -SYSTEM_PROGRAM = Pubkey.from_string("11111111111111111111111111111111") -SYSTEM_TOKEN_PROGRAM = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA") -SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM = Pubkey.from_string( +SYSTEM_PROGRAM: Final[Pubkey] = Pubkey.from_string("11111111111111111111111111111111") +SYSTEM_TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string( + "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" +) +SYSTEM_ASSOCIATED_TOKEN_ACCOUNT_PROGRAM: Final[Pubkey] = Pubkey.from_string( "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL" ) -SYSTEM_RENT = Pubkey.from_string("SysvarRent111111111111111111111111111111111") -SOL = Pubkey.from_string("So11111111111111111111111111111111111111112") -LAMPORTS_PER_SOL = 1_000_000_000 +SYSTEM_RENT: Final[Pubkey] = Pubkey.from_string( + "SysvarRent111111111111111111111111111111111" +) +SOL: Final[Pubkey] = Pubkey.from_string("So11111111111111111111111111111111111111112") +LAMPORTS_PER_SOL: Final[int] = 1_000_000_000 # Trading parameters BUY_AMOUNT = 0.0001 # Amount of SOL to spend when buying @@ -33,11 +49,12 @@ ENABLE_DYNAMIC_PRIORITY_FEE = ( ) EXTRA_PRIORITY_FEE = 0.1 # 10% increase in dynamic priority fee -# Your nodes -# You can also get a trader node https://docs.chainstack.com/docs/solana-trader-nodes -RPC_ENDPOINT = os.getenv("RPC_ENDPOINT") -WSS_ENDPOINT = os.getenv("WSS_ENDPOINT") -MAX_RPS = 25 # RPS of your node to avoid 429 errors +TOKEN_DECIMALS: Final[int] = 6 +MAX_RETRIES: int = 5 +WAIT_TIME_AFTER_BUY: int = 20 +WAIT_TIME_BEFORE_NEW_TOKEN: int = 5 +WAIT_TIME_AFTER_CREATION: int = 15 -# Private key -PRIVATE_KEY = os.getenv("PRIVATE_KEY") +# RPS of your node to avoid rate limit errors +# You can also get a trader node https://docs.chainstack.com/docs/solana-trader-nodes +MAX_RPS = 25 diff --git a/pyproject.toml b/pyproject.toml index d873eca..15adee2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,11 @@ -[tool.black] +[tool.ruff] line-length = 88 -target-version = ['py39'] +target-version = "py311" -[tool.isort] -profile = "black" \ No newline at end of file +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "N", "B", "A", "C4", "T10", "ARG", "PTH"] +ignore = ["E501"] + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" \ No newline at end of file diff --git a/src/core/__init__.py b/src/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/client.py b/src/core/client.py new file mode 100644 index 0000000..4f0335e --- /dev/null +++ b/src/core/client.py @@ -0,0 +1,162 @@ +""" +Solana client abstraction for blockchain operations. +""" + +import asyncio +from typing import Any, Dict, List, Optional, Tuple, Union + +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 Instruction +from solders.keypair import Keypair +from solders.pubkey import Pubkey + +from src.utils.logger import get_logger + +logger = get_logger(__name__) + + +class SolanaClient: + """Abstraction for Solana RPC client operations.""" + + def __init__(self, rpc_endpoint: str): + """Initialize Solana client with RPC endpoint. + + Args: + rpc_endpoint: URL of the Solana RPC endpoint + """ + self.rpc_endpoint = rpc_endpoint + self._client = None + + async def get_client(self) -> AsyncClient: + """Get or create the AsyncClient instance. + + Returns: + AsyncClient instance + """ + if self._client is None: + self._client = AsyncClient(self.rpc_endpoint) + return self._client + + async def close(self): + """Close the client connection if open.""" + if self._client: + await self._client.close() + self._client = None + + async def get_account_info(self, pubkey: Pubkey) -> Dict[str, Any]: + """Get account info from the blockchain. + + Args: + pubkey: Public key of the account + + Returns: + Account info response + + Raises: + ValueError: If account doesn't exist or has no data + """ + client = await self.get_client() + response = await client.get_account_info(pubkey) + if not response.value: + raise ValueError(f"Account {pubkey} not found") + return response.value + + async def get_token_account_balance(self, token_account: Pubkey) -> int: + """Get token balance for an account. + + Args: + token_account: Token account address + + Returns: + Token balance as integer + """ + client = await self.get_client() + response = await client.get_token_account_balance(token_account) + if response.value: + return int(response.value.amount) + return 0 + + async def get_latest_blockhash(self) -> str: + """Get the latest blockhash. + + Returns: + Recent blockhash as string + """ + client = await self.get_client() + response = await client.get_latest_blockhash() + return response.value.blockhash + + async def send_transaction( + self, + transaction: Transaction, + signer: Keypair, + skip_preflight: bool = True, + max_retries: int = 3, + ) -> str: + """Send a transaction to the network. + + Args: + transaction: Prepared transaction + signer: Transaction signer + skip_preflight: Whether to skip preflight checks + max_retries: Maximum number of sending attempts + + Returns: + Transaction signature + + Raises: + Exception: If transaction fails after all retries + """ + client = await self.get_client() + + # Ensure transaction has a recent blockhash + if not transaction.recent_blockhash: + blockhash = await self.get_latest_blockhash() + transaction.recent_blockhash = blockhash + + # Attempt to send with retries + for attempt in range(max_retries): + try: + tx_opts = TxOpts( + skip_preflight=skip_preflight, preflight_commitment=Confirmed + ) + response = await client.send_transaction( + transaction, signer, opts=tx_opts + ) + return response.value + + except Exception as e: + if attempt == max_retries - 1: + logger.error( + f"Failed to send transaction after {max_retries} attempts" + ) + raise + + wait_time = 2**attempt + logger.warning( + f"Transaction attempt {attempt + 1} failed: {str(e)}, retrying in {wait_time}s" + ) + await asyncio.sleep(wait_time) + + async def confirm_transaction( + self, signature: str, commitment: str = "confirmed" + ) -> bool: + """Wait for transaction confirmation. + + Args: + signature: Transaction signature + commitment: Confirmation commitment level + + Returns: + Whether transaction was confirmed + """ + client = await self.get_client() + try: + await client.confirm_transaction(signature, commitment=commitment) + return True + except Exception as e: + logger.error(f"Failed to confirm transaction {signature}: {str(e)}") + return False diff --git a/src/core/curve.py b/src/core/curve.py new file mode 100644 index 0000000..efca883 --- /dev/null +++ b/src/core/curve.py @@ -0,0 +1,136 @@ +""" +Bonding curve operations for pump.fun tokens. +""" + +import struct +from typing import Final + +from construct import Flag, Int64ul, Struct +from solana.rpc.async_api import AsyncClient +from solders.pubkey import Pubkey + +from src.core.client import SolanaClient +from src.core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS +from src.utils.logger import get_logger + +logger = get_logger(__name__) + +# Discriminator for the bonding curve account +EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" None: + """Parse bonding curve data. + + Args: + data: Raw account data + + Raises: + ValueError: If data cannot be parsed + """ + if data[:8] != EXPECTED_DISCRIMINATOR: + raise ValueError("Invalid curve state discriminator") + + parsed = self._STRUCT.parse(data[8:]) + self.__dict__.update(parsed) + + def calculate_price(self) -> float: + """Calculate token price in SOL. + + Returns: + Token price in SOL + + Raises: + ValueError: If reserve state is invalid + """ + if self.virtual_token_reserves <= 0 or self.virtual_sol_reserves <= 0: + raise ValueError("Invalid reserve state") + + return (self.virtual_sol_reserves / LAMPORTS_PER_SOL) / ( + self.virtual_token_reserves / 10**TOKEN_DECIMALS + ) + + @property + def token_reserves(self) -> float: + """Get token reserves in decimal form.""" + return self.virtual_token_reserves / 10**TOKEN_DECIMALS + + @property + def sol_reserves(self) -> float: + """Get SOL reserves in decimal form.""" + return self.virtual_sol_reserves / LAMPORTS_PER_SOL + + +class BondingCurveManager: + """Manager for bonding curve operations.""" + + def __init__(self, client: SolanaClient): + """Initialize with Solana client. + + Args: + client: Solana client for RPC calls + """ + self.client = client + + async def get_curve_state(self, curve_address: Pubkey) -> BondingCurveState: + """Get the state of a bonding curve. + + Args: + curve_address: Address of the bonding curve account + + Returns: + Bonding curve state + + Raises: + ValueError: If curve data is invalid + """ + try: + account = await self.client.get_account_info(curve_address) + if not account.data: + raise ValueError(f"No data in bonding curve account {curve_address}") + + return BondingCurveState(account.data) + + except Exception as e: + logger.error(f"Failed to get curve state: {str(e)}") + raise ValueError(f"Invalid curve state: {str(e)}") + + async def calculate_price(self, curve_address: Pubkey) -> float: + """Calculate the current price of a token. + + Args: + curve_address: Address of the bonding curve account + + Returns: + Token price in SOL + """ + curve_state = await self.get_curve_state(curve_address) + return curve_state.calculate_price() + + async def calculate_expected_tokens( + self, curve_address: Pubkey, sol_amount: float + ) -> float: + """Calculate the expected token amount for a given SOL input. + + Args: + curve_address: Address of the bonding curve account + sol_amount: Amount of SOL to spend + + Returns: + Expected token amount + """ + curve_state = await self.get_curve_state(curve_address) + price = curve_state.calculate_price() + return sol_amount / price diff --git a/src/core/pubkeys.py b/src/core/pubkeys.py new file mode 100644 index 0000000..7790ef7 --- /dev/null +++ b/src/core/pubkeys.py @@ -0,0 +1,51 @@ +""" +System and program addresses for Solana and pump.fun interactions. +""" + +from dataclasses import dataclass +from typing import Final + +from solders.pubkey import Pubkey + +LAMPORTS_PER_SOL: Final[int] = 1_000_000_000 +TOKEN_DECIMALS: Final[int] = 6 + + +@dataclass +class SystemAddresses: + """System-level Solana addresses.""" + + PROGRAM: Final[Pubkey] = Pubkey.from_string("11111111111111111111111111111111") + TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string( + "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" + ) + ASSOCIATED_TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string( + "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL" + ) + RENT: Final[Pubkey] = Pubkey.from_string( + "SysvarRent111111111111111111111111111111111" + ) + SOL: Final[Pubkey] = Pubkey.from_string( + "So11111111111111111111111111111111111111112" + ) + + +@dataclass +class PumpAddresses: + """Pump.fun program addresses.""" + + PROGRAM: Final[Pubkey] = Pubkey.from_string( + "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P" + ) + GLOBAL: Final[Pubkey] = Pubkey.from_string( + "4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf" + ) + EVENT_AUTHORITY: Final[Pubkey] = Pubkey.from_string( + "Ce6TQqeHC9p8KetsN6JsjHK7UTZk7nasjjnr7XxXp9F1" + ) + FEE: Final[Pubkey] = Pubkey.from_string( + "CebN5WGQ4jvEPvsVU4EoHEpgzq1VV7AbicfhtW4xC9iM" + ) + LIQUIDITY_MIGRATOR: Final[Pubkey] = Pubkey.from_string( + "39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg" + ) diff --git a/src/core/wallet.py b/src/core/wallet.py new file mode 100644 index 0000000..18c5956 --- /dev/null +++ b/src/core/wallet.py @@ -0,0 +1,55 @@ +""" +Wallet management for Solana transactions. +""" + +import base58 +from solders.keypair import Keypair +from solders.pubkey import Pubkey +from spl.token.instructions import get_associated_token_address + + +class Wallet: + """Manages a Solana wallet for trading operations.""" + + def __init__(self, private_key: str): + """Initialize wallet from private key. + + Args: + private_key: Base58 encoded private key + """ + self._private_key = private_key + self._keypair = self._load_keypair(private_key) + + @property + def pubkey(self) -> Pubkey: + """Get the public key of the wallet.""" + return self._keypair.pubkey() + + @property + def keypair(self) -> Keypair: + """Get the keypair for signing transactions.""" + return self._keypair + + def get_associated_token_address(self, mint: Pubkey) -> Pubkey: + """Get the associated token account address for a mint. + + Args: + mint: Token mint address + + Returns: + Associated token account address + """ + return get_associated_token_address(self.pubkey, mint) + + @staticmethod + def _load_keypair(private_key: str) -> Keypair: + """Load keypair from private key. + + Args: + private_key: Base58 encoded private key + + Returns: + Solana keypair + """ + private_key_bytes = base58.b58decode(private_key) + return Keypair.from_bytes(private_key_bytes) diff --git a/src/monitoring/__init__.py b/src/monitoring/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/monitoring/events.py b/src/monitoring/events.py new file mode 100644 index 0000000..d23c52d --- /dev/null +++ b/src/monitoring/events.py @@ -0,0 +1,168 @@ +""" +Event processing for pump.fun tokens. +""" + +import base64 +import json +import struct +from typing import Any, Callable, Dict, List, Optional + +from solders.pubkey import Pubkey +from solders.transaction import VersionedTransaction + +from src.trading.base import TokenInfo +from src.utils.logger import get_logger + +logger = get_logger(__name__) + + +class PumpEventProcessor: + """Processes events from pump.fun program.""" + + # Discriminator for create instruction + CREATE_DISCRIMINATOR = 8576854823835016728 + + def __init__(self, pump_program: Pubkey): + """Initialize event processor. + + Args: + pump_program: Pump.fun program address + """ + self.pump_program = pump_program + self._idl = self._load_idl() + + def _load_idl(self) -> Dict[str, Any]: + """Load IDL from file. + + Returns: + IDL as dictionary + """ + try: + with open("idl/pump_fun_idl.json", "r") as f: + return json.load(f) + except Exception as e: + logger.error(f"Failed to load IDL: {str(e)}") + # Create a minimal IDL with just what we need + return { + "instructions": [ + { + "name": "create", + "args": [ + {"name": "name", "type": "string"}, + {"name": "symbol", "type": "string"}, + {"name": "uri", "type": "string"}, + ], + } + ] + } + + def process_transaction(self, tx_data: str) -> Optional[TokenInfo]: + """Process a transaction and extract token info. + + Args: + tx_data: Base64 encoded transaction data + + Returns: + TokenInfo if a token creation is found, None otherwise + """ + try: + tx_data_decoded = base64.b64decode(tx_data) + transaction = VersionedTransaction.from_bytes(tx_data_decoded) + + for ix in transaction.message.instructions: + # Check if instruction is from pump.fun program + program_id_index = ix.program_id_index + if program_id_index >= len(transaction.message.account_keys): + continue + + program_id = transaction.message.account_keys[program_id_index] + + if str(program_id) != str(self.pump_program): + continue + + ix_data = bytes(ix.data) + + # Check if it's a create instruction + if len(ix_data) < 8: + continue + + discriminator = struct.unpack(" Dict[str, Any]: + """Decode create instruction data. + + Args: + ix_data: Instruction data bytes + ix_def: Instruction definition from IDL + accounts: List of account pubkeys + + Returns: + Decoded instruction arguments + """ + args = {} + offset = 8 # Skip 8-byte discriminator + + for arg in ix_def["args"]: + if arg["type"] == "string": + length = struct.unpack_from(" None: + """Listen for new token creations. + + Args: + token_callback: Callback function for new tokens + match_string: Optional string to match in token name/symbol + creator_address: Optional creator address to filter by + """ + while True: + try: + async with websockets.connect(self.wss_endpoint) as websocket: + await self._subscribe_to_program(websocket) + ping_task = asyncio.create_task(self._ping_loop(websocket)) + + try: + while True: + token_info = await self._wait_for_token_creation(websocket) + if not token_info: + continue + + logger.info( + f"New token detected: {token_info.name} ({token_info.symbol})" + ) + + # Apply filters + if match_string and not ( + match_string.lower() in token_info.name.lower() + or match_string.lower() in token_info.symbol.lower() + ): + logger.info( + f"Token does not match filter '{match_string}'. Skipping..." + ) + continue + + if ( + creator_address + and str(token_info.user) != creator_address + ): + logger.info( + f"Token not created by {creator_address}. Skipping..." + ) + continue + + await token_callback(token_info) + + except websockets.exceptions.ConnectionClosed: + logger.warning("WebSocket connection closed. Reconnecting...") + ping_task.cancel() + + except Exception as e: + logger.error(f"WebSocket connection error: {str(e)}") + logger.info("Reconnecting in 5 seconds...") + await asyncio.sleep(5) + + async def _subscribe_to_program(self, websocket) -> None: + """Subscribe to blocks mentioning the pump.fun program. + + Args: + websocket: Active WebSocket connection + """ + subscription_message = json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "method": "blockSubscribe", + "params": [ + {"mentionsAccountOrProgram": str(self.pump_program)}, + { + "commitment": "confirmed", + "encoding": "base64", + "showRewards": False, + "transactionDetails": "full", + "maxSupportedTransactionVersion": 0, + }, + ], + } + ) + + await websocket.send(subscription_message) + logger.info(f"Subscribed to blocks mentioning program: {self.pump_program}") + + async def _ping_loop(self, websocket) -> None: + """Keep connection alive with pings. + + Args: + websocket: Active WebSocket connection + """ + try: + while True: + await asyncio.sleep(self.ping_interval) + await websocket.ping() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Ping error: {str(e)}") + + async def _wait_for_token_creation(self, websocket) -> Optional[TokenInfo]: + """Wait for token creation event. + + Args: + websocket: Active WebSocket connection + + Returns: + TokenInfo if a token creation is found, None otherwise + """ + try: + response = await asyncio.wait_for(websocket.recv(), timeout=30) + data = json.loads(response) + + if "method" not in data or data["method"] != "blockNotification": + return None + + if "params" not in data or "result" not in data["params"]: + return None + + block_data = data["params"]["result"] + if "value" not in block_data or "block" not in block_data["value"]: + return None + + block = block_data["value"]["block"] + if "transactions" not in block: + return None + + for tx in block["transactions"]: + if not isinstance(tx, dict) or "transaction" not in tx: + continue + + token_info = self.event_processor.process_transaction( + tx["transaction"][0] + ) + if token_info: + return token_info + + except asyncio.TimeoutError: + logger.debug("No data received for 30 seconds") + except websockets.exceptions.ConnectionClosed: + logger.warning("WebSocket connection closed") + raise + except Exception as e: + logger.error(f"Error processing WebSocket message: {str(e)}") + + return None diff --git a/src/trading/__init__.py b/src/trading/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/trading/base.py b/src/trading/base.py new file mode 100644 index 0000000..e6fd6b5 --- /dev/null +++ b/src/trading/base.py @@ -0,0 +1,83 @@ +""" +Base interfaces for trading operations. +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Tuple + +from solders.pubkey import Pubkey +from solders.signature import Signature + + +@dataclass +class TokenInfo: + """Token information.""" + + name: str + symbol: str + uri: str + mint: Pubkey + bonding_curve: Pubkey + associated_bonding_curve: Pubkey + user: Pubkey + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "TokenInfo": + """Create TokenInfo from dictionary. + + Args: + data: Dictionary with token data + + Returns: + TokenInfo instance + """ + return cls( + name=data["name"], + symbol=data["symbol"], + uri=data["uri"], + mint=Pubkey.from_string(data["mint"]), + bonding_curve=Pubkey.from_string(data["bondingCurve"]), + associated_bonding_curve=Pubkey.from_string(data["associatedBondingCurve"]), + user=Pubkey.from_string(data["user"]), + ) + + def to_dict(self) -> Dict[str, str]: + """Convert to dictionary. + + Returns: + Dictionary representation + """ + return { + "name": self.name, + "symbol": self.symbol, + "uri": self.uri, + "mint": str(self.mint), + "bondingCurve": str(self.bonding_curve), + "associatedBondingCurve": str(self.associated_bonding_curve), + "user": str(self.user), + } + + +@dataclass +class TradeResult: + """Result of a trading operation.""" + + success: bool + tx_signature: Optional[str] = None + error_message: Optional[str] = None + amount: Optional[float] = None + price: Optional[float] = None + + +class Trader(ABC): + """Base interface for trading operations.""" + + @abstractmethod + async def execute(self, *args, **kwargs) -> TradeResult: + """Execute trading operation. + + Returns: + TradeResult with operation outcome + """ + pass diff --git a/src/trading/buyer.py b/src/trading/buyer.py new file mode 100644 index 0000000..52dd329 --- /dev/null +++ b/src/trading/buyer.py @@ -0,0 +1,253 @@ +""" +Buy operations for pump.fun tokens. +""" + +import asyncio +import struct +from typing import List, Optional + +import spl.token.instructions as spl_token +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 spl.token.instructions import get_associated_token_address + +from src.core.client import SolanaClient +from src.core.curve import BondingCurveManager +from src.core.pubkeys import ( + LAMPORTS_PER_SOL, + TOKEN_DECIMALS, + PumpAddresses, + SystemAddresses, +) +from src.core.wallet import Wallet +from src.trading.base import TokenInfo, Trader, TradeResult +from src.utils.logger import get_logger + +logger = get_logger(__name__) + + +class TokenBuyer(Trader): + """Handles buying tokens on pump.fun.""" + + def __init__( + self, + client: SolanaClient, + wallet: Wallet, + curve_manager: BondingCurveManager, + amount: float, + slippage: float = 0.01, + max_retries: int = 5, + ): + """Initialize token buyer. + + Args: + client: Solana client for RPC calls + wallet: Wallet for signing transactions + curve_manager: Bonding curve manager + amount: Amount of SOL to spend + slippage: Slippage tolerance (0.01 = 1%) + max_retries: Maximum number of retry attempts + """ + self.client = client + self.wallet = wallet + self.curve_manager = curve_manager + self.amount = amount + self.slippage = slippage + self.max_retries = max_retries + + async def execute(self, token_info: TokenInfo, *args, **kwargs) -> TradeResult: + """Execute buy operation. + + Args: + token_info: Token information + + Returns: + TradeResult with buy outcome + """ + try: + # Extract token info + mint = token_info.mint + bonding_curve = token_info.bonding_curve + associated_bonding_curve = token_info.associated_bonding_curve + + # Convert amount to lamports + amount_lamports = int(self.amount * LAMPORTS_PER_SOL) + + # Fetch token price + curve_state = await self.curve_manager.get_curve_state(bonding_curve) + token_price_sol = curve_state.calculate_price() + token_amount = self.amount / token_price_sol + + # Calculate maximum SOL to spend with slippage + max_amount_lamports = int(amount_lamports * (1 + self.slippage)) + + logger.info( + f"Buying {token_amount:.6f} tokens at {token_price_sol:.8f} SOL per token" + ) + logger.info( + f"Total cost: {self.amount:.6f} SOL (max: {max_amount_lamports / LAMPORTS_PER_SOL:.6f} SOL)" + ) + + associated_token_account = self.wallet.get_associated_token_address(mint) + + await self._ensure_associated_token_account(mint, associated_token_account) + + tx_signature = await self._send_buy_transaction( + mint, + bonding_curve, + associated_bonding_curve, + associated_token_account, + token_amount, + max_amount_lamports, + ) + + success = await self.client.confirm_transaction(tx_signature) + + if success: + logger.info(f"Buy transaction confirmed: {tx_signature}") + return TradeResult( + success=True, + tx_signature=tx_signature, + amount=token_amount, + price=token_price_sol, + ) + else: + return TradeResult( + success=False, + error_message=f"Transaction failed to confirm: {tx_signature}", + ) + + except Exception as e: + logger.error(f"Buy operation failed: {str(e)}") + return TradeResult(success=False, error_message=str(e)) + + async def _ensure_associated_token_account( + self, mint: Pubkey, associated_token_account: Pubkey + ) -> None: + """Ensure associated token account exists. + + Args: + mint: Token mint + associated_token_account: Associated token account address + """ + try: + solana_client = await self.client.get_client() + account_info = await solana_client.get_account_info( + associated_token_account + ) + + if account_info.value is None: + logger.info(f"Creating associated token account for {mint}...") + + create_ata_ix = spl_token.create_associated_token_account( + payer=self.wallet.pubkey, owner=self.wallet.pubkey, mint=mint + ) + + create_ata_tx = Transaction() + create_ata_tx.add(create_ata_ix) + blockhash = await self.client.get_latest_blockhash() + create_ata_tx.recent_blockhash = blockhash + + tx_sig = await self.client.send_transaction( + create_ata_tx, self.wallet.keypair + ) + + await self.client.confirm_transaction(tx_sig) + logger.info( + f"Associated token account created: {associated_token_account}" + ) + else: + logger.info( + f"Associated token account already exists: {associated_token_account}" + ) + + except Exception as e: + logger.error(f"Error creating associated token account: {str(e)}") + raise + + async def _send_buy_transaction( + self, + mint: Pubkey, + bonding_curve: Pubkey, + associated_bonding_curve: Pubkey, + associated_token_account: Pubkey, + token_amount: float, + max_amount_lamports: int, + ) -> str: + """Send buy transaction. + + Args: + mint: Token mint + bonding_curve: Bonding curve address + associated_bonding_curve: Associated bonding curve address + associated_token_account: User's token account + token_amount: Amount of tokens to buy + max_amount_lamports: Maximum SOL to spend in lamports + + Returns: + Transaction signature + + Raises: + Exception: If transaction fails after all retries + """ + accounts = [ + AccountMeta( + pubkey=PumpAddresses.GLOBAL, is_signer=False, is_writable=False + ), + AccountMeta(pubkey=PumpAddresses.FEE, is_signer=False, is_writable=True), + AccountMeta(pubkey=mint, 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=associated_token_account, is_signer=False, is_writable=True + ), + AccountMeta(pubkey=self.wallet.pubkey, is_signer=True, is_writable=True), + AccountMeta( + pubkey=SystemAddresses.PROGRAM, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=SystemAddresses.RENT, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=PumpAddresses.EVENT_AUTHORITY, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=PumpAddresses.PROGRAM, is_signer=False, is_writable=False + ), + ] + + # Prepare buy instruction data + # Discriminator for buy instruction + discriminator = struct.pack(" TradeResult: + """Execute sell operation. + + Args: + token_info: Token information + + Returns: + TradeResult with sell outcome + """ + try: + # Extract token info + mint = token_info.mint + bonding_curve = token_info.bonding_curve + associated_bonding_curve = token_info.associated_bonding_curve + + # Get associated token account + associated_token_account = self.wallet.get_associated_token_address(mint) + + # Get token balance + token_balance = await self.client.get_token_account_balance( + associated_token_account + ) + token_balance_decimal = token_balance / 10**TOKEN_DECIMALS + + logger.info(f"Token balance: {token_balance_decimal}") + + if token_balance == 0: + logger.info("No tokens to sell.") + return TradeResult(success=False, error_message="No tokens to sell") + + # Fetch token price + curve_state = await self.curve_manager.get_curve_state(bonding_curve) + token_price_sol = curve_state.calculate_price() + + logger.info(f"Price per Token: {token_price_sol:.8f} SOL") + + # Calculate minimum SOL output with slippage + amount = token_balance + expected_sol_output = float(token_balance_decimal) * float(token_price_sol) + slippage_factor = 1 - self.slippage + min_sol_output = int( + (expected_sol_output * slippage_factor) * LAMPORTS_PER_SOL + ) + + logger.info(f"Selling {token_balance_decimal} tokens") + logger.info(f"Expected SOL output: {expected_sol_output:.8f} SOL") + logger.info( + f"Minimum SOL output (with {self.slippage * 100}% slippage): {min_sol_output / LAMPORTS_PER_SOL:.8f} SOL" + ) + + tx_signature = await self._send_sell_transaction( + mint, + bonding_curve, + associated_bonding_curve, + associated_token_account, + amount, + min_sol_output, + ) + + success = await self.client.confirm_transaction(tx_signature) + + if success: + logger.info(f"Sell transaction confirmed: {tx_signature}") + return TradeResult( + success=True, + tx_signature=tx_signature, + amount=token_balance_decimal, + price=token_price_sol, + ) + else: + return TradeResult( + success=False, + error_message=f"Transaction failed to confirm: {tx_signature}", + ) + + except Exception as e: + logger.error(f"Sell operation failed: {str(e)}") + return TradeResult(success=False, error_message=str(e)) + + async def _send_sell_transaction( + self, + mint: Pubkey, + bonding_curve: Pubkey, + associated_bonding_curve: Pubkey, + associated_token_account: Pubkey, + token_amount: int, + min_sol_output: int, + ) -> str: + """Send sell transaction. + + Args: + mint: Token mint + bonding_curve: Bonding curve address + associated_bonding_curve: Associated bonding curve address + associated_token_account: User's token account + token_amount: Amount of tokens to sell in raw units + min_sol_output: Minimum SOL to receive in lamports + + Returns: + Transaction signature + + Raises: + Exception: If transaction fails after all retries + """ + # Prepare sell instruction accounts + accounts = [ + AccountMeta( + pubkey=PumpAddresses.GLOBAL, is_signer=False, is_writable=False + ), + AccountMeta(pubkey=PumpAddresses.FEE, is_signer=False, is_writable=True), + AccountMeta(pubkey=mint, 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=associated_token_account, is_signer=False, is_writable=True + ), + AccountMeta(pubkey=self.wallet.pubkey, is_signer=True, is_writable=True), + AccountMeta( + pubkey=SystemAddresses.PROGRAM, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=SystemAddresses.ASSOCIATED_TOKEN_PROGRAM, + is_signer=False, + is_writable=False, + ), + AccountMeta( + pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=PumpAddresses.EVENT_AUTHORITY, is_signer=False, is_writable=False + ), + AccountMeta( + pubkey=PumpAddresses.PROGRAM, is_signer=False, is_writable=False + ), + ] + + # Prepare sell instruction data + # Discriminator for sell instruction + discriminator = struct.pack(" None: + """Start the trading bot. + + Args: + match_string: Optional string to match in token name/symbol + bro_address: Optional creator address to filter by + marry_mode: If True, only buy tokens and skip selling + yolo_mode: If True, trade continuously + """ + logger.info("Starting pump.fun trader") + logger.info(f"Match filter: {match_string if match_string else 'None'}") + logger.info(f"Creator filter: {bro_address if bro_address else 'None'}") + logger.info(f"Marry mode: {marry_mode}") + logger.info(f"YOLO mode: {yolo_mode}") + + try: + await self.token_listener.listen_for_tokens( + lambda token: self._handle_new_token(token, marry_mode, yolo_mode), + match_string, + bro_address, + ) + + except Exception as e: + logger.error(f"Trading stopped due to error: {str(e)}") + await self.solana_client.close() + + async def _handle_new_token( + self, token_info: TokenInfo, marry_mode: bool, yolo_mode: bool + ) -> None: + """Handle a new token creation event. + + Args: + token_info: Token information + marry_mode: If True, only buy tokens and skip selling + yolo_mode: If True, continue trading after this token + """ + try: + await self._save_token_info(token_info) + + logger.info("Waiting for 15 seconds for the bonding curve to stabilize...") + await asyncio.sleep(15) + + try: + token_price = await self.curve_manager.calculate_price( + token_info.bonding_curve + ) + logger.info(f"Token price: {token_price:.10f} SOL") + except Exception as e: + logger.error(f"Failed to get token price: {str(e)}") + token_price = 0 + + logger.info( + f"Buying {self.buy_amount:.6f} SOL worth of {token_info.symbol}..." + ) + buy_result = await self.buyer.execute(token_info) + + if buy_result.success: + logger.info(f"Successfully bought {token_info.symbol}") + self._log_trade("buy", token_info, token_price, buy_result.tx_signature) + else: + logger.error( + f"Failed to buy {token_info.symbol}: {buy_result.error_message}" + ) + + # Sell token if not in marry mode + if not marry_mode and buy_result.success: + logger.info("Waiting for 20 seconds before selling...") + await asyncio.sleep(20) + + logger.info(f"Selling {token_info.symbol}...") + sell_result = await self.seller.execute(token_info) + + if sell_result.success: + logger.info(f"Successfully sold {token_info.symbol}") + self._log_trade( + "sell", token_info, token_price, sell_result.tx_signature + ) + else: + logger.error( + f"Failed to sell {token_info.symbol}: {sell_result.error_message}" + ) + elif marry_mode: + logger.info("Marry mode enabled. Skipping sell operation.") + + # Wait before looking for the next token + if yolo_mode: + logger.info( + "YOLO mode enabled. Waiting 5 seconds before looking for next token..." + ) + await asyncio.sleep(5) + + except Exception as e: + logger.error(f"Error handling token {token_info.symbol}: {str(e)}") + + async def _save_token_info(self, token_info: TokenInfo) -> None: + """Save token information to a file. + + Args: + token_info: Token information + """ + os.makedirs("trades", exist_ok=True) + file_name = os.path.join("trades", f"{token_info.mint}.txt") + + with open(file_name, "w") as file: + file.write(json.dumps(token_info.to_dict(), indent=2)) + + logger.info(f"Token information saved to {file_name}") + + def _log_trade( + self, action: str, token_info: TokenInfo, price: float, tx_hash: str | None + ) -> None: + """Log trade information. + + Args: + action: Trade action (buy/sell) + token_info: Token information + price: Token price in SOL + tx_hash: Transaction hash + """ + os.makedirs("trades", exist_ok=True) + + log_entry = { + "timestamp": datetime.utcnow().isoformat(), + "action": action, + "token_address": str(token_info.mint), + "symbol": token_info.symbol, + "price": price, + "tx_hash": tx_hash, + } + + with open("trades/trades.log", "a") as log_file: + log_file.write(json.dumps(log_entry) + "\n") diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils/logger.py b/src/utils/logger.py new file mode 100644 index 0000000..35dabd2 --- /dev/null +++ b/src/utils/logger.py @@ -0,0 +1,65 @@ +""" +Logging utilities for the pump.fun trading bot. +""" + +import logging +import sys +from typing import Dict + +# Global dict to store loggers +_loggers: Dict[str, logging.Logger] = {} + + +def get_logger(name: str, level: int = logging.INFO) -> logging.Logger: + """Get or create a logger with the given name. + + Args: + name: Logger name, typically __name__ + level: Logging level + + Returns: + Configured logger + """ + global _loggers + + if name in _loggers: + return _loggers[name] + + logger = logging.getLogger(name) + logger.setLevel(level) + + if not logger.handlers: + formatter = logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + _loggers[name] = logger + return logger + + +def setup_file_logging( + filename: str = "pump_trading.log", level: int = logging.INFO +) -> None: + """Set up file logging for all loggers. + + Args: + filename: Log file path + level: Logging level for file handler + """ + root_logger = logging.getLogger() + + formatter = logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + file_handler = logging.FileHandler(filename) + file_handler.setLevel(level) + file_handler.setFormatter(formatter) + + root_logger.addHandler(file_handler)