From fce547ec9e142fdd9c50e15d38ce6b9bca2c776c Mon Sep 17 00:00:00 2001 From: smypmsa Date: Thu, 6 Mar 2025 16:14:17 +0000 Subject: [PATCH] fixed and buy transactions --- .gitignore | 5 ++- cli.py | 23 +++-------- config.py | 14 +++---- src/core/client.py | 14 ++----- src/monitoring/listener.py | 9 ++++- src/trading/buyer.py | 37 +++++++++--------- src/trading/seller.py | 20 ++++++---- src/trading/trader.py | 78 ++++++++++++++++++++++++++++++++++++-- 8 files changed, 133 insertions(+), 67 deletions(-) diff --git a/.gitignore b/.gitignore index 55804a2..13525c6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ +trades/* + .pylintrc .ruff_cache @@ -162,4 +164,5 @@ cython_debug/ # be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ \ No newline at end of file +#.idea/ +trades/trades.log diff --git a/cli.py b/cli.py index 2701092..5923dde 100644 --- a/cli.py +++ b/cli.py @@ -52,16 +52,7 @@ def parse_args() -> argparse.Namespace: 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() @@ -73,23 +64,19 @@ async def main() -> None: # Get configuration values, preferring command line args over config.py rpc_endpoint: str | None = ( - args.rpc - or os.environ.get("SOLANA_NODE_RPC_ENDPOINT") - or config.PUBLIC_RPC_ENDPOINT + os.environ.get("SOLANA_NODE_RPC_ENDPOINT") or config.PUBLIC_RPC_ENDPOINT ) wss_endpoint: str | None = ( - args.wss - or os.environ.get("SOLANA_NODE_WSS_ENDPOINT") - or config.PUBLIC_WSS_ENDPOINT + os.environ.get("SOLANA_NODE_WSS_ENDPOINT") or config.PUBLIC_WSS_ENDPOINT ) - private_key: str | None = args.key or os.environ.get("SOLANA_PRIVATE_KEY") + private_key: str | None = os.environ.get("SOLANA_PRIVATE_KEY") # Validate configuration values - if not rpc_endpoint or rpc_endpoint.startswith(("http://", "https://")): + if not rpc_endpoint or not 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://")): + if not wss_endpoint or not wss_endpoint.startswith(("ws://", "wss://")): logger.error("Invalid WebSocket endpoint. Must start with ws:// or wss://") sys.exit(1) diff --git a/config.py b/config.py index 36c273d..168b997 100644 --- a/config.py +++ b/config.py @@ -3,17 +3,17 @@ Configuration for the pump.fun trading bot. """ # Trading parameters -BUY_AMOUNT = 0.0001 # Amount of SOL to spend when buying -BUY_SLIPPAGE = 0.2 # 20% slippage tolerance for buying -SELL_SLIPPAGE = 0.2 # 20% slippage tolerance for selling +BUY_AMOUNT = 0.000001 # Amount of SOL to spend when buying +BUY_SLIPPAGE = 0.4 # 40% slippage tolerance for buying +SELL_SLIPPAGE = 0.4 # 40% slippage tolerance for selling ENABLE_DYNAMIC_PRIORITY_FEE = True # TODO: not implemented. getRecentPriorityFee is used to get current priority fee EXTRA_PRIORITY_FEE = 0.1 # TODO: not implemented. 10% increase in dynamic priority fee # Retries and timeouts -MAX_RETRIES: int = 5 -WAIT_TIME_AFTER_BUY: int = 5 -WAIT_TIME_BEFORE_NEW_TOKEN: int = 5 -WAIT_TIME_AFTER_CREATION: int = 5 +MAX_RETRIES: int = 2 +WAIT_TIME_AFTER_BUY: int = 15 +WAIT_TIME_BEFORE_NEW_TOKEN: int = 120 +WAIT_TIME_AFTER_CREATION: int = 15 # Node provier configuration # You can also get a trader node https://docs.chainstack.com/docs/solana-trader-nodes diff --git a/src/core/client.py b/src/core/client.py index 7f9ea2c..159b908 100644 --- a/src/core/client.py +++ b/src/core/client.py @@ -8,6 +8,7 @@ from typing import Any, Dict from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Confirmed from solana.rpc.types import TxOpts +from solders.hash import Hash from solders.keypair import Keypair from solders.pubkey import Pubkey from solders.transaction import Transaction @@ -78,7 +79,7 @@ class SolanaClient: return int(response.value.amount) return 0 - async def get_latest_blockhash(self) -> str: + async def get_latest_blockhash(self) -> Hash: """Get the latest blockhash. Returns: @@ -91,7 +92,6 @@ class SolanaClient: async def send_transaction( self, transaction: Transaction, - signer: Keypair, skip_preflight: bool = True, max_retries: int = 3, ) -> str: @@ -99,7 +99,6 @@ class SolanaClient: Args: transaction: Prepared transaction - signer: Transaction signer skip_preflight: Whether to skip preflight checks max_retries: Maximum number of sending attempts @@ -111,20 +110,13 @@ class SolanaClient: """ 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 - ) + response = await client.send_transaction(transaction, opts=tx_opts) return response.value except Exception as e: diff --git a/src/monitoring/listener.py b/src/monitoring/listener.py index b529243..1b29028 100644 --- a/src/monitoring/listener.py +++ b/src/monitoring/listener.py @@ -125,7 +125,14 @@ class PumpTokenListener: try: while True: await asyncio.sleep(self.ping_interval) - await websocket.ping() + try: + pong_waiter = await websocket.ping() + await asyncio.wait_for(pong_waiter, timeout=10) + except asyncio.TimeoutError: + logger.warning("Ping timeout - server not responding") + # Force reconnection + await websocket.close() + return except asyncio.CancelledError: pass except Exception as e: diff --git a/src/trading/buyer.py b/src/trading/buyer.py index d85aa89..a26e27e 100644 --- a/src/trading/buyer.py +++ b/src/trading/buyer.py @@ -4,18 +4,20 @@ Buy operations for pump.fun tokens. import asyncio import struct -from typing import List, Optional +from typing import Final, 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 solders.hash import Hash 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 -from spl.token.instructions import get_associated_token_address +from spl.token.instructions import create_associated_token_account from src.core.client import SolanaClient from src.core.curve import BondingCurveManager @@ -31,6 +33,9 @@ from src.utils.logger import get_logger logger = get_logger(__name__) +# Discriminator for the buy instruction +EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" None: + """Queue a token for processing if not already processed.""" + token_key = str(token_info.mint) + + if token_key in self.processed_tokens: + logger.debug(f"Token {token_info.symbol} already processed. Skipping...") + return + + # Record timestamp when token was discovered + self.token_timestamps[token_key] = asyncio.get_event_loop().time() + + # Add to queue + await self.token_queue.put(token_info) + logger.info(f"Queued new token: {token_info.symbol} ({token_info.mint})") + + async def _process_token_queue(self, marry_mode: bool, yolo_mode: bool) -> None: + """Continuously process tokens from the queue, only if they're fresh.""" + while True: + # Get next token + token_info = await self.token_queue.get() + token_key = str(token_info.mint) + + # Check if token is still fresh + current_time = asyncio.get_event_loop().time() + token_age = current_time - self.token_timestamps.get( + token_key, current_time + ) + + if token_age > self.max_token_age: + logger.info( + f"Skipping token {token_info.symbol} - too old ({token_age:.1f}s > {self.max_token_age}s)" + ) + self.token_queue.task_done() + continue + + # Mark as processing + self.processed_tokens.add(token_key) + + # Process token + logger.info( + f"Processing fresh token: {token_info.symbol} (age: {token_age:.1f}s)" + ) + await self._handle_token(token_info, marry_mode, yolo_mode) + + # Mark queue task as done + self.token_queue.task_done() + + async def _handle_token( self, token_info: TokenInfo, marry_mode: bool, yolo_mode: bool ) -> None: """Handle a new token creation event. @@ -168,6 +231,13 @@ class PumpTrader: elif marry_mode: logger.info("Marry mode enabled. Skipping sell operation.") + # Wait before looking for the next token + if yolo_mode: + logger.info( + f"YOLO mode enabled. Waiting {config.WAIT_TIME_BEFORE_NEW_TOKEN} seconds before looking for next token..." + ) + await asyncio.sleep(config.WAIT_TIME_BEFORE_NEW_TOKEN) + except Exception as e: logger.error(f"Error handling token {token_info.symbol}: {str(e)}") @@ -211,7 +281,7 @@ class PumpTrader: "symbol": token_info.symbol, "price": price, "amount": amount, - "tx_hash": tx_hash, + "tx_hash": str(tx_hash), } with open("trades/trades.log", "a") as log_file: