gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-07-27 23:37:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(pump): fix curve management

Browse Source
This commit is contained in:
smypmsa
2025-08-02 14:17:29 +00:00
parent 8ee6b95d2d
commit 04342fee5a
2 changed files with 167 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/curve.py
-165
View File
@@ -1,165 +0,0 @@
"""
Pump.fun bonding curve manager for price calculations and curve state management.
"""
import struct
from dataclasses import dataclass
from solders.pubkey import Pubkey
from core.client import SolanaClient
from core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS
from utils.logger import get_logger
logger = get_logger(__name__)
# Bonding curve discriminator
CURVE_DISCRIMINATOR = bytes([23, 203, 71, 8, 209, 70, 227, 3])
@dataclass
class BondingCurveState:
"""Represents the state of a pump.fun bonding curve."""
virtual_token_reserves: int
virtual_sol_reserves: int
real_token_reserves: int
real_sol_reserves: int
token_total_supply: int
complete: bool
creator: Pubkey
@property
def token_reserves(self) -> float:
"""Token reserves in decimal form."""
return self.virtual_token_reserves / 10**TOKEN_DECIMALS
@property
def sol_reserves(self) -> float:
"""SOL reserves in decimal form."""
return self.virtual_sol_reserves / LAMPORTS_PER_SOL
def calculate_price(self) -> float:
"""Calculate current token price in SOL."""
if self.virtual_token_reserves <= 0:
return 0.0
# Price = sol_reserves / token_reserves
price_lamports = self.virtual_sol_reserves / self.virtual_token_reserves
return price_lamports * (10**TOKEN_DECIMALS) / LAMPORTS_PER_SOL
class BondingCurveManager:
"""Manages pump.fun bonding curve operations."""
def __init__(self, client: SolanaClient):
"""Initialize bonding curve manager.
Args:
client: Solana RPC client
"""
self.client = client
async def get_curve_state(self, curve_address: Pubkey) -> BondingCurveState:
"""Get the current state of a bonding curve.
Args:
curve_address: Address of the bonding curve
Returns:
BondingCurveState with current curve data
Raises:
ValueError: If curve data is invalid or inaccessible
"""
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}")
curve_state = self._decode_curve_state(account.data)
return curve_state
except Exception as e:
logger.error(f"Failed to get curve state: {e!s}")
raise ValueError(f"Invalid bonding curve state: {e!s}")
async def calculate_price(self, curve_address: Pubkey) -> float:
"""Calculate current token price from bonding curve.
Args:
curve_address: Address of the bonding curve
Returns:
Current token price in SOL
"""
curve_state = await self.get_curve_state(curve_address)
return curve_state.calculate_price()
def _decode_curve_state(self, data: bytes) -> BondingCurveState:
"""Decode bonding curve state from raw account data.
Args:
data: Raw account data
Returns:
Decoded BondingCurveState
Raises:
ValueError: If data format is invalid
"""
if len(data) < 8:
raise ValueError("Curve data too short")
# Check discriminator
if not data.startswith(CURVE_DISCRIMINATOR):
raise ValueError("Invalid curve discriminator")
offset = 8
try:
# Decode based on pump.fun BondingCurve structure:
# - virtual_token_reserves: u64 (8 bytes)
# - virtual_sol_reserves: u64 (8 bytes)
# - real_token_reserves: u64 (8 bytes)
# - real_sol_reserves: u64 (8 bytes)
# - token_total_supply: u64 (8 bytes)
# - complete: bool (1 byte)
# - padding: 7 bytes
# - creator: Pubkey (32 bytes)
virtual_token_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
virtual_sol_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
real_token_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
real_sol_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
token_total_supply = struct.unpack_from("<Q", data, offset)[0]
offset += 8
complete = bool(struct.unpack_from("<B", data, offset)[0])
offset += 1
# Skip padding
offset += 7
creator = Pubkey.from_bytes(data[offset:offset + 32])
return BondingCurveState(
virtual_token_reserves=virtual_token_reserves,
virtual_sol_reserves=virtual_sol_reserves,
real_token_reserves=real_token_reserves,
real_sol_reserves=real_sol_reserves,
token_total_supply=token_total_supply,
complete=complete,
creator=creator,
)
except Exception as e:
raise ValueError(f"Failed to decode curve state: {e}")
src/platforms/pumpfun/curve_manager.py
+167 -28
View File
@@ -2,17 +2,56 @@
Pump.Fun implementation of CurveManager interface.
This module handles pump.fun-specific bonding curve operations
by implementing the CurveManager interface.
by implementing the CurveManager interface with all curve logic self-contained.
"""
import struct
from dataclasses import dataclass
from typing import Any
from solders.pubkey import Pubkey
from core.client import SolanaClient
from core.curve import BondingCurveManager
from core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS
from interfaces.core import CurveManager, Platform
from utils.logger import get_logger
logger = get_logger(__name__)
# Bonding curve discriminator for pump.fun
CURVE_DISCRIMINATOR = bytes([23, 203, 71, 8, 209, 70, 227, 3])
@dataclass
class BondingCurveState:
"""Represents the state of a pump.fun bonding curve."""
virtual_token_reserves: int
virtual_sol_reserves: int
real_token_reserves: int
real_sol_reserves: int
token_total_supply: int
complete: bool
creator: Pubkey
@property
def token_reserves(self) -> float:
"""Token reserves in decimal form."""
return self.virtual_token_reserves / 10**TOKEN_DECIMALS
@property
def sol_reserves(self) -> float:
"""SOL reserves in decimal form."""
return self.virtual_sol_reserves / LAMPORTS_PER_SOL
def calculate_price(self) -> float:
"""Calculate current token price in SOL."""
if self.virtual_token_reserves <= 0:
return 0.0
# Price = sol_reserves / token_reserves
price_lamports = self.virtual_sol_reserves / self.virtual_token_reserves
return price_lamports * (10**TOKEN_DECIMALS) / LAMPORTS_PER_SOL
class PumpFunCurveManager(CurveManager):
@@ -25,7 +64,6 @@ class PumpFunCurveManager(CurveManager):
client: Solana RPC client
"""
self.client = client
self.bonding_curve_manager = BondingCurveManager(client)
@property
def platform(self) -> Platform:
@@ -41,22 +79,31 @@ class PumpFunCurveManager(CurveManager):
Returns:
Dictionary containing bonding curve state data
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
return {
"virtual_token_reserves": curve_state.virtual_token_reserves,
"virtual_sol_reserves": curve_state.virtual_sol_reserves,
"real_token_reserves": curve_state.real_token_reserves,
"real_sol_reserves": curve_state.real_sol_reserves,
"token_total_supply": curve_state.token_total_supply,
"complete": curve_state.complete,
"creator": str(curve_state.creator),
try:
account = await self.client.get_account_info(pool_address)
if not account.data:
raise ValueError(f"No data in bonding curve account {pool_address}")
# Calculated fields for convenience
"price_per_token": curve_state.calculate_price(),
"token_reserves_decimal": curve_state.token_reserves,
"sol_reserves_decimal": curve_state.sol_reserves,
}
curve_state = self._decode_curve_state(account.data)
return {
"virtual_token_reserves": curve_state.virtual_token_reserves,
"virtual_sol_reserves": curve_state.virtual_sol_reserves,
"real_token_reserves": curve_state.real_token_reserves,
"real_sol_reserves": curve_state.real_sol_reserves,
"token_total_supply": curve_state.token_total_supply,
"complete": curve_state.complete,
"creator": str(curve_state.creator),
# Calculated fields for convenience
"price_per_token": curve_state.calculate_price(),
"token_reserves_decimal": curve_state.token_reserves,
"sol_reserves_decimal": curve_state.sol_reserves,
}
except Exception as e:
logger.error(f"Failed to get curve state: {e!s}")
raise ValueError(f"Invalid bonding curve state: {e!s}")
async def calculate_price(self, pool_address: Pubkey) -> float:
"""Calculate current token price from bonding curve state.
@@ -67,7 +114,8 @@ class PumpFunCurveManager(CurveManager):
Returns:
Current token price in SOL
"""
return await self.bonding_curve_manager.calculate_price(pool_address)
curve_state = await self._get_curve_state_object(pool_address)
return curve_state.calculate_price()
async def calculate_buy_amount_out(
self,
@@ -85,7 +133,7 @@ class PumpFunCurveManager(CurveManager):
Returns:
Expected amount of tokens to receive (in raw token units)
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
curve_state = await self._get_curve_state_object(pool_address)
# Use virtual reserves for bonding curve calculation
# Formula: tokens_out = (amount_in * virtual_token_reserves) / (virtual_sol_reserves + amount_in)
@@ -114,7 +162,7 @@ class PumpFunCurveManager(CurveManager):
Returns:
Expected amount of SOL to receive (in lamports)
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
curve_state = await self._get_curve_state_object(pool_address)
# Use virtual reserves for bonding curve calculation
# Formula: sol_out = (amount_in * virtual_sol_reserves) / (virtual_token_reserves + amount_in)
@@ -136,14 +184,106 @@ class PumpFunCurveManager(CurveManager):
Returns:
Tuple of (token_reserves, sol_reserves) in raw units
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
curve_state = await self._get_curve_state_object(pool_address)
return (curve_state.virtual_token_reserves, curve_state.virtual_sol_reserves)
async def _get_curve_state_object(self, curve_address: Pubkey) -> BondingCurveState:
"""Get the current state of a bonding curve as a BondingCurveState object.
Args:
curve_address: Address of the bonding curve
Returns:
BondingCurveState with current curve data
Raises:
ValueError: If curve data is invalid or inaccessible
"""
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}")
curve_state = self._decode_curve_state(account.data)
return curve_state
except Exception as e:
logger.error(f"Failed to get curve state: {e!s}")
raise ValueError(f"Invalid bonding curve state: {e!s}")
def _decode_curve_state(self, data: bytes) -> BondingCurveState:
"""Decode bonding curve state from raw account data.
Args:
data: Raw account data
Returns:
Decoded BondingCurveState
Raises:
ValueError: If data format is invalid
"""
if len(data) < 8:
raise ValueError("Curve data too short")
# Check discriminator
if not data.startswith(CURVE_DISCRIMINATOR):
raise ValueError("Invalid curve discriminator")
offset = 8
try:
# Decode based on pump.fun BondingCurve structure:
# - virtual_token_reserves: u64 (8 bytes)
# - virtual_sol_reserves: u64 (8 bytes)
# - real_token_reserves: u64 (8 bytes)
# - real_sol_reserves: u64 (8 bytes)
# - token_total_supply: u64 (8 bytes)
# - complete: bool (1 byte)
# - padding: 7 bytes
# - creator: Pubkey (32 bytes)
virtual_token_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
virtual_sol_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
real_token_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
real_sol_reserves = struct.unpack_from("<Q", data, offset)[0]
offset += 8
token_total_supply = struct.unpack_from("<Q", data, offset)[0]
offset += 8
complete = bool(struct.unpack_from("<B", data, offset)[0])
offset += 1
# Skip padding
offset += 7
creator = Pubkey.from_bytes(data[offset:offset + 32])
return BondingCurveState(
virtual_token_reserves=virtual_token_reserves,
virtual_sol_reserves=virtual_sol_reserves,
real_token_reserves=real_token_reserves,
real_sol_reserves=real_sol_reserves,
token_total_supply=token_total_supply,
complete=complete,
creator=creator,
)
except Exception as e:
raise ValueError(f"Failed to decode curve state: {e}")
# Additional convenience methods for pump.fun specific operations
async def calculate_expected_tokens(self, pool_address: Pubkey, sol_amount: float) -> float:
"""Calculate the expected token amount for a given SOL input.
This is a convenience method that wraps calculate_buy_amount_out
and converts between decimal and raw units.
This is a convenience method that converts between decimal and raw units.
Args:
pool_address: Address of the bonding curve
@@ -159,8 +299,7 @@ class PumpFunCurveManager(CurveManager):
async def calculate_expected_sol(self, pool_address: Pubkey, token_amount: float) -> float:
"""Calculate the expected SOL amount for a given token input.
This is a convenience method that wraps calculate_sell_amount_out
and converts between decimal and raw units.
This is a convenience method that converts between decimal and raw units.
Args:
pool_address: Address of the bonding curve
@@ -182,7 +321,7 @@ class PumpFunCurveManager(CurveManager):
Returns:
True if curve is complete, False otherwise
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
curve_state = await self._get_curve_state_object(pool_address)
return curve_state.complete
async def get_curve_progress(self, pool_address: Pubkey) -> dict[str, Any]:
@@ -194,7 +333,7 @@ class PumpFunCurveManager(CurveManager):
Returns:
Dictionary with progress information
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
curve_state = await self._get_curve_state_object(pool_address)
# Calculate progress based on SOL raised vs target
# This is approximate since the exact target isn't stored in the curve state