gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-08-17 01:08:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs(claude): add claude code rules

Browse Source
This commit is contained in:
smypmsa
2025-08-11 05:35:25 +00:00
parent f2efd56f81
commit 8ab8932168
56 changed files with 4193 additions and 2605 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/platforms/letsbonk/curve_manager.py
+72 -71
View File
@@ -21,10 +21,10 @@ logger = get_logger(__name__)
class LetsBonkCurveManager(CurveManager):
"""LetsBonk (Raydium LaunchLab) implementation of CurveManager interface."""
def __init__(self, client: SolanaClient, idl_parser: IDLParser):
"""Initialize LetsBonk curve manager with injected IDL parser.
Args:
client: Solana RPC client
idl_parser: Pre-loaded IDL parser for LetsBonk platform
@@ -32,20 +32,20 @@ class LetsBonkCurveManager(CurveManager):
self.client = client
self.address_provider = LetsBonkAddressProvider()
self._idl_parser = idl_parser
logger.info("LetsBonk curve manager initialized with injected IDL parser")
@property
def platform(self) -> Platform:
"""Get the platform this manager serves."""
return Platform.LETS_BONK
async def get_pool_state(self, pool_address: Pubkey) -> dict[str, Any]:
"""Get the current state of a LetsBonk pool.
Args:
pool_address: Address of the pool state account
Returns:
Dictionary containing pool state data
"""
@@ -53,136 +53,130 @@ class LetsBonkCurveManager(CurveManager):
account = await self.client.get_account_info(pool_address)
if not account.data:
raise ValueError(f"No data in pool state account {pool_address}")
# Decode pool state using injected IDL parser
pool_state_data = self._decode_pool_state_with_idl(account.data)
return pool_state_data
except Exception as e:
logger.exception("Failed to get pool state")
raise ValueError(f"Invalid pool state: {e!s}")
async def calculate_price(self, pool_address: Pubkey) -> float:
"""Calculate current token price from pool state.
Args:
pool_address: Address of the pool state
Returns:
Current token price in SOL
"""
pool_state = await self.get_pool_state(pool_address)
# Use virtual reserves for price calculation
virtual_base = pool_state["virtual_base"]
virtual_quote = pool_state["virtual_quote"]
if virtual_base <= 0 or virtual_quote <= 0:
raise ValueError("Invalid reserve state")
# Price = quote_reserves / base_reserves (how much SOL per token)
price_lamports = virtual_quote / virtual_base
price_sol = price_lamports * (10**TOKEN_DECIMALS) / LAMPORTS_PER_SOL
return price_sol
async def calculate_buy_amount_out(
self,
pool_address: Pubkey,
amount_in: int
self, pool_address: Pubkey, amount_in: int
) -> int:
"""Calculate expected tokens received for a buy operation.
Uses the constant product AMM formula.
Args:
pool_address: Address of the pool state
amount_in: Amount of SOL to spend (in lamports)
Returns:
Expected amount of tokens to receive (in raw token units)
"""
pool_state = await self.get_pool_state(pool_address)
virtual_base = pool_state["virtual_base"]
virtual_quote = pool_state["virtual_quote"]
# Constant product formula: tokens_out = (amount_in * virtual_base) / (virtual_quote + amount_in)
numerator = amount_in * virtual_base
denominator = virtual_quote + amount_in
if denominator == 0:
return 0
tokens_out = numerator // denominator
return tokens_out
async def calculate_sell_amount_out(
self,
pool_address: Pubkey,
amount_in: int
self, pool_address: Pubkey, amount_in: int
) -> int:
"""Calculate expected SOL received for a sell operation.
Uses the constant product AMM formula.
Args:
pool_address: Address of the pool state
amount_in: Amount of tokens to sell (in raw token units)
Returns:
Expected amount of SOL to receive (in lamports)
"""
pool_state = await self.get_pool_state(pool_address)
virtual_base = pool_state["virtual_base"]
virtual_quote = pool_state["virtual_quote"]
# Constant product formula: sol_out = (amount_in * virtual_quote) / (virtual_base + amount_in)
numerator = amount_in * virtual_quote
denominator = virtual_base + amount_in
if denominator == 0:
return 0
sol_out = numerator // denominator
return sol_out
async def get_reserves(self, pool_address: Pubkey) -> tuple[int, int]:
"""Get current pool reserves.
Args:
pool_address: Address of the pool state
Returns:
Tuple of (base_reserves, quote_reserves) in raw units
"""
pool_state = await self.get_pool_state(pool_address)
return (pool_state["virtual_base"], pool_state["virtual_quote"])
def _decode_pool_state_with_idl(self, data: bytes) -> dict[str, Any]:
"""Decode pool state data using injected IDL parser.
Args:
data: Raw account data
Returns:
Dictionary with decoded pool state
Raises:
ValueError: If IDL parsing fails
"""
# Use injected IDL parser to decode PoolState account data
decoded_pool_state = self._idl_parser.decode_account_data(
data,
"PoolState",
skip_discriminator=True
data, "PoolState", skip_discriminator=True
)
if not decoded_pool_state:
raise ValueError("Failed to decode pool state with IDL parser")
# Extract the fields we need for trading calculations
# Based on the PoolState structure from the IDL
pool_data = {
@@ -193,28 +187,31 @@ class LetsBonkCurveManager(CurveManager):
"status": decoded_pool_state.get("status", 0),
"supply": decoded_pool_state.get("supply", 0),
}
# Calculate additional metrics
if pool_data["virtual_base"] > 0:
pool_data["price_per_token"] = (
(pool_data["virtual_quote"] / pool_data["virtual_base"])
* (10**TOKEN_DECIMALS) / LAMPORTS_PER_SOL
(pool_data["virtual_quote"] / pool_data["virtual_base"])
* (10**TOKEN_DECIMALS)
/ LAMPORTS_PER_SOL
)
else:
pool_data["price_per_token"] = 0
logger.debug(f"Decoded pool state: virtual_base={pool_data['virtual_base']}, "
f"virtual_quote={pool_data['virtual_quote']}, "
f"price={pool_data['price_per_token']:.8f} SOL")
logger.debug(
f"Decoded pool state: virtual_base={pool_data['virtual_base']}, "
f"virtual_quote={pool_data['virtual_quote']}, "
f"price={pool_data['price_per_token']:.8f} SOL"
)
return pool_data
async def validate_pool_state_structure(self, pool_address: Pubkey) -> bool:
"""Validate that the pool state structure matches IDL expectations.
Args:
pool_address: Address of the pool state
Returns:
True if structure is valid, False otherwise
"""
@@ -222,23 +219,27 @@ class LetsBonkCurveManager(CurveManager):
# This would be used during development/testing to ensure
# the IDL parsing is working correctly
pool_state = await self.get_pool_state(pool_address)
required_fields = [
"virtual_base", "virtual_quote",
"real_base", "real_quote"
"virtual_base",
"virtual_quote",
"real_base",
"real_quote",
]
for field in required_fields:
if field not in pool_state:
logger.error(f"Missing required field: {field}")
return False
if not isinstance(pool_state[field], int):
logger.error(f"Field {field} is not an integer: {type(pool_state[field])}")
logger.error(
f"Field {field} is not an integer: {type(pool_state[field])}"
)
return False
return True
except Exception:
logger.exception("Pool state validation failed")
return False
return False