gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-11-02 06:16:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(cookbook,tools): fail on an unreadable value instead of substituting a default (#225)

Browse Source 
* fix(cookbook,tools): read GEYSER_AUTH_TYPE and reject an unknown scheme

Four Geyser clients advertised the variable and ignored it, sending Basic for
anything that was not exactly "x-token". Validate against {x-token, basic}.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): detect blockSubscribe creations from logs alone

The envelope fallback cannot see a create reached by CPI, so it covered a
subset of the log route. Envelope decoding stays in the decode/ script.
Also skip meta.err: logs emitted before a failure are kept, and that mint
does not exist.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): raise when the graduation baseline cannot be read

The watchers fell back to a hardcoded 793,100,000 and still printed it as
"(from Global)". Every progress figure is measured against that baseline.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): report an unsent create flag as unknown, not false

Truncated create_v2 args and legacy create were filled in as False, reading
as a coin the program said is not in mayhem mode. Leave them out, print N/A.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(tools): build Geyser credentials before the reconnect loop

The loop catches every exception and sleeps, so an unusable auth type
reconnected forever instead of reporting a misconfiguration.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): treat a short Global buffer as an unreadable baseline

The guard only caught a missing account, so a short read raised struct.error
past the documented ValueError.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(tools): surface a listener's exception instead of an idle market

A listener that raised on connect returned nothing, exactly like a quiet
market, in the harness whose job is to certify listeners. Take the listener
list from the factory too, so one added there is not silently skipped.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(tools): simulate the transaction the bot built

The hook rebuilt the transaction itself and dropped the account data size
limit, so the simulation was more permissive than the send. Hook the RPC
client instead, build the client before detection so its blockhash cache is
warm, and report the compute budget read back off the wire.

A listener that raised on connect also read as a quiet market here.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(tools): skip failed migrations on err, not on the word "Error"

The notification carries err beside the logs. Scanning the text dropped a
migration whose logs merely said "Error" and kept ones that truly failed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): refuse a letsbonk buy that accepts any output

The only one of the four without the guard. Near graduation the floor
division reaches zero, and the buy submits with no slippage protection.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): require the pool's own platform_config

The fallback was a partner-specific address that reverts with
ConstraintAddress (2012). A pool that omits it cannot be traded.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): raise when curve data ends before a flag

is_mayhem_mode picks the fee recipient for a buy, and the wrong one reverts
with NotAuthorized (6000), so defaulting it only moves the failure.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cookbook): trade the coin's own PumpSwap pool, not an arbitrary one

Anyone can open a pool against any mint and getProgramAccounts has no defined
order, so [0] was a stranger's pool. Select on coin_creator, which migration
copies from the bonding curve and nothing else sets, and refuse rather than
guess when there is no single match. --pool trades a specific one.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Anton Sauchyk
2026-09-24 17:49:12 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent c00cd608e3
commit ae937fd477
22 changed files with 562 additions and 524 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cookbook/letsbonk/letsbonk_buy_token_exact_in.py
+14 -13
View File
@@ -60,13 +60,6 @@ RAYDIUM_LAUNCHLAB_PROGRAM_ID = Pubkey.from_string(
"LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj"
)
GLOBAL_CONFIG = Pubkey.from_string("6s1xP3hpbAfFoNtUNF8mfHsjr2Bd97JxFJRWLbL6aHuX")
# Fallback only. platform_config is NOT the same for every LaunchLab pool:
# partner launches use their own, and passing the wrong one fails the buy/sell
# with ConstraintAddress (2012). The live value is read from the pool state
# below; this constant is only used if the pool omits it.
LETSBONK_PLATFORM_CONFIG = Pubkey.from_string(
"5thqcDwKp5QQ8US4XRMoseGeGbmLKMmoKZmS6zHrQAsA"
)
# Token program and system addresses
TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
@@ -454,12 +447,13 @@ async def buy_exact_in(
# Derive necessary PDAs
authority = derive_authority_pda()
event_authority = derive_event_authority_pda()
# platform_config varies per pool — take the pool's own value.
platform_config = (
Pubkey.from_string(pool_state_data["platform_config"])
if pool_state_data.get("platform_config")
else LETSBONK_PLATFORM_CONFIG
)
# platform_config varies per pool: partner launches use their own, and
# the wrong one fails the trade with ConstraintAddress (2012). There is
# no value to fall back to, so a pool that omits it cannot be traded.
if not pool_state_data.get("platform_config"):
print("Pool state carries no platform_config; cannot build the trade")
return None
platform_config = Pubkey.from_string(pool_state_data["platform_config"])
creator_fee_vault = derive_creator_fee_vault(creator, WSOL_MINT)
platform_fee_vault = derive_platform_fee_vault(
platform_config, WSOL_MINT
@@ -474,6 +468,13 @@ async def buy_exact_in(
pool_state_data, amount_in, slippage_tolerance
)
# expected_output is a floor division by the quote reserve, so once the
# pool nears graduation an ordinary amount_in floors to zero and the
# buy would submit accepting any output at all.
if minimum_amount_out is None or minimum_amount_out == 0:
print("Failed to calculate minimum amount out or amount is too small")
return None
print(f"Amount in: {amount_in} lamports ({amount_in_sol} SOL)")
print(f"Minimum amount out: {minimum_amount_out}")
cookbook/letsbonk/letsbonk_buy_token_exact_out.py
+7 -13
View File
@@ -60,13 +60,6 @@ RAYDIUM_LAUNCHLAB_PROGRAM_ID = Pubkey.from_string(
"LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj"
)
GLOBAL_CONFIG = Pubkey.from_string("6s1xP3hpbAfFoNtUNF8mfHsjr2Bd97JxFJRWLbL6aHuX")
# Fallback only. platform_config is NOT the same for every LaunchLab pool:
# partner launches use their own, and passing the wrong one fails the buy/sell
# with ConstraintAddress (2012). The live value is read from the pool state
# below; this constant is only used if the pool omits it.
LETSBONK_PLATFORM_CONFIG = Pubkey.from_string(
"5thqcDwKp5QQ8US4XRMoseGeGbmLKMmoKZmS6zHrQAsA"
)
# Token program and system addresses
TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
@@ -459,12 +452,13 @@ async def buy_exact_out(
# Derive necessary PDAs
authority = derive_authority_pda()
event_authority = derive_event_authority_pda()
# platform_config varies per pool — take the pool's own value.
platform_config = (
Pubkey.from_string(pool_state_data["platform_config"])
if pool_state_data.get("platform_config")
else LETSBONK_PLATFORM_CONFIG
)
# platform_config varies per pool: partner launches use their own, and
# the wrong one fails the trade with ConstraintAddress (2012). There is
# no value to fall back to, so a pool that omits it cannot be traded.
if not pool_state_data.get("platform_config"):
print("Pool state carries no platform_config; cannot build the trade")
return None
platform_config = Pubkey.from_string(pool_state_data["platform_config"])
creator_fee_vault = derive_creator_fee_vault(creator, WSOL_MINT)
platform_fee_vault = derive_platform_fee_vault(platform_config, WSOL_MINT)
cookbook/letsbonk/letsbonk_sell_token_exact_in.py
+7 -13
View File
@@ -60,13 +60,6 @@ RAYDIUM_LAUNCHLAB_PROGRAM_ID = Pubkey.from_string(
"LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj"
)
GLOBAL_CONFIG = Pubkey.from_string("6s1xP3hpbAfFoNtUNF8mfHsjr2Bd97JxFJRWLbL6aHuX")
# Fallback only. platform_config is NOT the same for every LaunchLab pool:
# partner launches use their own, and passing the wrong one fails the buy/sell
# with ConstraintAddress (2012). The live value is read from the pool state
# below; this constant is only used if the pool omits it.
LETSBONK_PLATFORM_CONFIG = Pubkey.from_string(
"5thqcDwKp5QQ8US4XRMoseGeGbmLKMmoKZmS6zHrQAsA"
)
# Token program and system addresses
TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
@@ -456,12 +449,13 @@ async def sell_exact_in(
# Derive necessary PDAs
authority = derive_authority_pda()
event_authority = derive_event_authority_pda()
# platform_config varies per pool — take the pool's own value.
platform_config = (
Pubkey.from_string(pool_state_data["platform_config"])
if pool_state_data.get("platform_config")
else LETSBONK_PLATFORM_CONFIG
)
# platform_config varies per pool: partner launches use their own, and
# the wrong one fails the trade with ConstraintAddress (2012). There is
# no value to fall back to, so a pool that omits it cannot be traded.
if not pool_state_data.get("platform_config"):
print("Pool state carries no platform_config; cannot build the trade")
return None
platform_config = Pubkey.from_string(pool_state_data["platform_config"])
creator_fee_vault = derive_creator_fee_vault(creator, WSOL_MINT)
platform_fee_vault = derive_platform_fee_vault(platform_config, WSOL_MINT)
cookbook/letsbonk/letsbonk_sell_token_exact_out.py
+7 -13
View File
@@ -60,13 +60,6 @@ RAYDIUM_LAUNCHLAB_PROGRAM_ID = Pubkey.from_string(
"LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj"
)
GLOBAL_CONFIG = Pubkey.from_string("6s1xP3hpbAfFoNtUNF8mfHsjr2Bd97JxFJRWLbL6aHuX")
# Fallback only. platform_config is NOT the same for every LaunchLab pool:
# partner launches use their own, and passing the wrong one fails the buy/sell
# with ConstraintAddress (2012). The live value is read from the pool state
# below; this constant is only used if the pool omits it.
LETSBONK_PLATFORM_CONFIG = Pubkey.from_string(
"5thqcDwKp5QQ8US4XRMoseGeGbmLKMmoKZmS6zHrQAsA"
)
# Token program and system addresses
TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
@@ -459,12 +452,13 @@ async def sell_exact_out(
# Derive necessary PDAs
authority = derive_authority_pda()
event_authority = derive_event_authority_pda()
# platform_config varies per pool — take the pool's own value.
platform_config = (
Pubkey.from_string(pool_state_data["platform_config"])
if pool_state_data.get("platform_config")
else LETSBONK_PLATFORM_CONFIG
)
# platform_config varies per pool: partner launches use their own, and
# the wrong one fails the trade with ConstraintAddress (2012). There is
# no value to fall back to, so a pool that omits it cannot be traded.
if not pool_state_data.get("platform_config"):
print("Pool state carries no platform_config; cannot build the trade")
return None
platform_config = Pubkey.from_string(pool_state_data["platform_config"])
creator_fee_vault = derive_creator_fee_vault(creator, WSOL_MINT)
platform_fee_vault = derive_platform_fee_vault(
platform_config, WSOL_MINT
cookbook/pumpfun/decode/pumpfun_decode_transaction_gettransaction.py
+11 -9
View File
@@ -116,7 +116,6 @@ def decode_create_instruction(data: bytes) -> dict:
"uri": results[2],
"creator": creator,
"token_standard": "legacy",
"is_mayhem_mode": False,
}
@@ -147,21 +146,24 @@ def decode_create_v2_instruction(data: bytes) -> dict:
)
offset += 32
is_mayhem_mode = bool(data[offset]) if offset < len(data) else False
offset += 1
is_cashback_enabled = bool(data[offset]) if offset < len(data) else False
return {
decoded = {
"name": results[0],
"symbol": results[1],
"uri": results[2],
"creator": creator,
"token_standard": "token2022",
"is_mayhem_mode": is_mayhem_mode,
"is_cashback_enabled": is_cashback_enabled,
}
# Both trailing args may be cut off the wire. A missing one is left out, so
# the caller can tell "not sent" from a flag that was sent as false.
if offset < len(data):
decoded["is_mayhem_mode"] = bool(data[offset])
offset += 1
if offset < len(data):
decoded["is_cashback_enabled"] = bool(data[offset])
return decoded
def decode_trade_instruction(data: bytes) -> dict:
"""Decode the two u64 args shared by buy/sell and their v2 forms.
cookbook/pumpfun/graduation/pumpfun_watch_graduating_geyser.py
+21 -13
View File
@@ -122,8 +122,9 @@ QUOTE_SYMBOLS: Final[dict[Pubkey, str]] = {WSOL_MINT: "SOL", USDC_MINT: "USDC"}
# Same offset in SPL Token and Token-2022: extensions are appended after it.
_MINT_DECIMALS_OFFSET: Final[int] = 44
# Only used if the Global account cannot be read: 1B supply less 206.9M reserved.
FALLBACK_INITIAL_REAL_TOKEN_RESERVES: Final[float] = 793_100_000.0
_NO_BASELINE_MSG: Final[str] = (
"Cannot read initial_real_token_reserves from the pump.fun Global account"
)
# A qualifying curve is traded several times a second. Reprint it only once it has
# moved this far, so the output stays readable.
@@ -297,18 +298,25 @@ async def fetch_initial_real_token_reserves(client: AsyncClient) -> float:
client: Connected RPC client
Returns:
Initial real token reserves in whole tokens, or the fallback constant
Initial real token reserves in whole tokens
Raises:
ValueError: If Global is missing or carries a zero at that offset. The
baseline is the 0% mark every progress figure is measured against,
so a guessed one misreports every coin the run touches.
"""
try:
resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64")
raw = struct.unpack_from(
"<Q", resp.value.data, _GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET
)[0]
if raw:
return raw / 10**TOKEN_DECIMALS
except Exception as e: # noqa: BLE001 - fall back rather than abort the watcher
print(f"⚠️ Could not read Global, using the fallback baseline: {e}")
return FALLBACK_INITIAL_REAL_TOKEN_RESERVES
resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64")
if (
resp.value is None
or len(resp.value.data) < _GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET + 8
):
raise ValueError(_NO_BASELINE_MSG)
raw = struct.unpack_from(
"<Q", resp.value.data, _GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET
)[0]
if not raw:
raise ValueError(_NO_BASELINE_MSG)
return raw / 10**TOKEN_DECIMALS
async def resolve_mint(client: AsyncClient, curve: Pubkey) -> Pubkey | None:
cookbook/pumpfun/graduation/pumpfun_watch_graduating_programsubscribe.py
+21 -13
View File
@@ -117,8 +117,9 @@ QUOTE_SYMBOLS: Final[dict[Pubkey, str]] = {WSOL_MINT: "SOL", USDC_MINT: "USDC"}
# Same offset in SPL Token and Token-2022: extensions are appended after it.
_MINT_DECIMALS_OFFSET: Final[int] = 44
# Only used if the Global account cannot be read: 1B supply less 206.9M reserved.
FALLBACK_INITIAL_REAL_TOKEN_RESERVES: Final[float] = 793_100_000.0
_NO_BASELINE_MSG: Final[str] = (
"Cannot read initial_real_token_reserves from the pump.fun Global account"
)
# A qualifying curve is traded several times a second. Reprint it only once it has
# moved this far, so the output stays readable.
@@ -256,18 +257,25 @@ async def fetch_initial_real_token_reserves(client: AsyncClient) -> float:
client: Connected RPC client
Returns:
Initial real token reserves in whole tokens, or the fallback constant
Initial real token reserves in whole tokens
Raises:
ValueError: If Global is missing or carries a zero at that offset. The
baseline is the 0% mark every progress figure is measured against,
so a guessed one misreports every coin the run touches.
"""
try:
resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64")
raw = struct.unpack_from(
"<Q", resp.value.data, _GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET
)[0]
if raw:
return raw / 10**TOKEN_DECIMALS
except Exception as e: # noqa: BLE001 - fall back rather than abort the watcher
print(f"⚠️ Could not read Global, using the fallback baseline: {e}")
return FALLBACK_INITIAL_REAL_TOKEN_RESERVES
resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64")
if (
resp.value is None
or len(resp.value.data) < _GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET + 8
):
raise ValueError(_NO_BASELINE_MSG)
raw = struct.unpack_from(
"<Q", resp.value.data, _GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET
)[0]
if not raw:
raise ValueError(_NO_BASELINE_MSG)
return raw / 10**TOKEN_DECIMALS
async def resolve_mint(client: AsyncClient, curve: Pubkey) -> Pubkey | None:
cookbook/pumpfun/listen/pumpfun_listen_tokens_blocksubscribe.py
+36 -275
View File
@@ -1,5 +1,6 @@
"""Listens to Solana blocks for Pump.fun 'create' instructions via WebSocket.
Decodes transaction data to extract mint, bonding curve, and user details.
"""Listens to Solana blocks for Pump.fun coin creations via WebSocket.
Reads each transaction's `meta.logMessages` for the CreateEvent the program
emits, which carries the mint, bonding curve and creator as literal pubkeys.
Usage:
uv run cookbook/pumpfun/listen/pumpfun_listen_tokens_blocksubscribe.py
@@ -7,8 +8,10 @@ Usage:
Performance: Usually slower than other listeners due to block-level processing.
This script uses blockSubscribe which receives entire blocks containing transactions
that mention the Pump.fun program. It then decodes the instruction data from each
transaction to extract token creation details.
that mention the Pump.fun program. Detection routes on the logs rather than the
transaction envelope: a `create` reached by CPI is in the logs and absent from the
envelope's top-level instructions. To take an envelope apart, see
cookbook/pumpfun/decode/pumpfun_decode_transaction_blocksubscribe.py.
WebSocket API Reference:
https://solana.com/docs/rpc/websocket/blocksubscribe
@@ -23,11 +26,9 @@ import json
import os
import struct
import base58
import websockets
from dotenv import load_dotenv
from solders.pubkey import Pubkey
from solders.transaction import VersionedTransaction
load_dotenv()
@@ -38,21 +39,11 @@ WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
# instead of delivering the message. Same value the bot's own listeners use.
WEBSOCKET_MAX_MESSAGE_BYTES = 32 * 1024 * 1024
PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
TOKEN_PROGRAM = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
TOKEN_2022_PROGRAM = Pubkey.from_string("TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb")
ASSOCIATED_TOKEN_PROGRAM = Pubkey.from_string(
"ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL"
)
# Instruction discriminators (8-byte identifiers for instruction types)
# Calculated using the first 8 bytes of sha256("global:create") for legacy Create
# and sha256("global:createV2") for Token2022 CreateV2
# See: cookbook/solana/anchor_calculate_discriminator.py
CREATE_DISCRIMINATOR = 8576854823835016728
CREATE_V2_DISCRIMINATOR = struct.unpack(
"<Q", bytes([214, 144, 76, 236, 95, 139, 49, 180])
)[0]
def print_token_info(token_data, signature=None):
"""Print token information in a consistent, user-friendly format.
@@ -78,7 +69,7 @@ def print_token_info(token_data, signature=None):
print(f"Creator: {token_data['creator']}")
print(f"Token Standard: {token_data.get('token_standard', 'N/A')}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', False)}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', 'N/A')}")
if "uri" in token_data:
print(f"URI: {token_data['uri']}")
@@ -88,48 +79,6 @@ def print_token_info(token_data, signature=None):
print("=" * 80 + "\n")
def get_account_keys(transaction, instruction, loaded_addresses=None):
"""Safely extract account keys for an instruction from a versioned transaction.
Handles both static account keys and loaded addresses from lookup tables.
Args:
loaded_addresses: Dict with 'writable' and 'readonly' loaded addresses from tx meta
Returns:
List of account keys as strings, or None if unable to resolve
"""
account_keys = []
static_keys = transaction.message.account_keys
# Combine all available account keys: static + loaded
all_keys = list(static_keys)
if loaded_addresses:
if "writable" in loaded_addresses:
for addr in loaded_addresses["writable"]:
all_keys.append(Pubkey.from_string(addr))
if "readonly" in loaded_addresses:
for addr in loaded_addresses["readonly"]:
all_keys.append(Pubkey.from_string(addr))
# Now resolve account indices
for index in instruction.accounts:
try:
if index < len(all_keys):
account_keys.append(str(all_keys[index]))
else:
print(
f"Warning: Account index {index} out of range (max: {len(all_keys) - 1})"
)
return None
except (IndexError, Exception) as e:
print(f"Error resolving account at index {index}: {e}")
return None
return account_keys
# First 8 bytes of sha256("event:CreateEvent"). Anchor emits the event as a
# base64 "Program data:" log line, which the RPC has already decoded for us.
CREATE_EVENT_DISCRIMINATOR = bytes([27, 114, 169, 77, 222, 235, 99, 118])
@@ -244,224 +193,39 @@ def find_create_event(logs):
return None
def load_idl(file_path):
with open(file_path) as f:
return json.load(f)
def decode_create_instruction(ix_data, ix_def, accounts):
"""Decode legacy Create instruction (Metaplex tokens).
The Create instruction creates tokens using the Metaplex Token Metadata standard.
Instruction data contains: name, symbol, uri, and additional creator pubkey.
Account references are extracted from the accounts array.
Args:
ix_data: Raw instruction data bytes
ix_def: Instruction definition from IDL
accounts: List of account pubkeys involved in the instruction
Returns:
Dictionary containing decoded token information
"""
args = {}
offset = 8 # Skip 8-byte discriminator
# Parse instruction arguments according to IDL definition
for arg in ix_def["args"]:
if arg["type"] == "string":
# String format: 4-byte length prefix + UTF-8 encoded string
length = struct.unpack_from("<I", ix_data, offset)[0]
offset += 4
value = ix_data[offset : offset + length].decode("utf-8")
offset += length
elif arg["type"] == "pubkey":
# Pubkey is 32 bytes, encoded as base58
value = base58.b58encode(ix_data[offset : offset + 32]).decode("utf-8")
offset += 32
else:
raise ValueError(f"Unsupported type: {arg['type']}")
args[arg["name"]] = value
# Extract account addresses from the accounts array
# Account layout for Create instruction:
# 0: mint, 1: metadata, 2: bondingCurve, 3: associatedBondingCurve,
# 4: tokenProgram, 5: systemProgram, 6: rent, 7: user
args["mint"] = str(accounts[0])
args["bondingCurve"] = str(accounts[2])
args["associatedBondingCurve"] = str(accounts[3])
args["user"] = str(accounts[7])
args["token_standard"] = "legacy"
args["is_mayhem_mode"] = False
return args
def decode_create_v2_instruction(ix_data, ix_def, accounts):
"""Decode CreateV2 instruction (Token2022 tokens).
The CreateV2 instruction creates tokens using the Token-2022 standard, which supports
additional features like transfer fees, interest-bearing tokens, and more.
This instruction includes an optional is_mayhem_mode flag.
Token-2022 Reference:
https://spl.solana.com/token-2022
Args:
ix_data: Raw instruction data bytes
ix_def: Instruction definition from IDL
accounts: List of account pubkeys involved in the instruction
Returns:
Dictionary containing decoded token information
"""
args = {}
offset = 8 # Skip 8-byte discriminator
# create_v2 args: name, symbol, uri, creator (pubkey), is_mayhem_mode (bool),
# is_cashback_enabled (OptionBool), creator_fee_bps (OptionU64),
# is_holder_reward (OptionBool).
#
# OptionBool and OptionU64 are single-field Anchor structs with no presence
# tag: each serializes as its bare inner value, 1 and 8 bytes. They are
# positional rather than independently optional, and the trailing ones are
# legally absent from the wire — three lengths occur on chain (no trailing
# args, is_cashback_enabled only, is_cashback_enabled plus creator_fee_bps).
# An absent arg is reported as None, meaning unset, not as a fabricated
# default; reading a fixed number of trailing bytes raises IndexError.
for arg in ix_def["args"]:
t = arg["type"]
if t == "string":
length = struct.unpack_from("<I", ix_data, offset)[0]
offset += 4
value = ix_data[offset : offset + length].decode("utf-8")
offset += length
elif t == "pubkey":
value = base58.b58encode(ix_data[offset : offset + 32]).decode("utf-8")
offset += 32
elif t == "bool":
value = bool(ix_data[offset]) if offset < len(ix_data) else False
offset += 1
elif isinstance(t, dict) and "defined" in t:
defined_name = (
t["defined"]["name"] if isinstance(t["defined"], dict) else t["defined"]
)
if defined_name == "OptionBool":
if offset >= len(ix_data):
value = None
else:
value = bool(ix_data[offset])
offset += 1
elif defined_name == "OptionU64":
if offset + 8 > len(ix_data):
value = None
else:
value = struct.unpack_from("<Q", ix_data, offset)[0]
offset += 8
else:
raise ValueError(f"Unsupported defined type: {defined_name}")
else:
raise ValueError(f"Unsupported type: {t}")
args[arg["name"]] = value
# Extract account addresses from the accounts array
# Account layout for CreateV2 instruction:
# 0: mint, 1: metadata, 2: bondingCurve, 3: associatedBondingCurve,
# 4: tokenProgram (Token2022), 5: user, 6: systemProgram, 7: rent
args["mint"] = str(accounts[0])
args["bondingCurve"] = str(accounts[2])
args["associatedBondingCurve"] = str(accounts[3])
args["user"] = str(accounts[5])
args["token_standard"] = "token2022"
return args
def decode_from_envelope(tx, idl):
"""Decode a coin creation straight from the transaction bytes.
The fallback for a block that arrives without `logMessages`. It is a
fallback and not the main path because the envelope is the one part of a
transaction whose format changes underneath you: solders 0.26 raises
`ValueError: io error: unexpected end of file` on a v1 transaction, and
solders only learned to read one in 0.29, which needs solana-py 0.40.
Args:
tx: One entry from a blockSubscribe notification's `transactions`
idl: The parsed pump.fun IDL
Returns:
True if a creation was found and printed
"""
try:
transaction = VersionedTransaction.from_bytes(
base64.b64decode(tx["transaction"][0])
)
except (ValueError, KeyError, IndexError):
return False
meta = tx.get("meta") or {}
loaded_addresses = meta.get("loadedAddresses")
for ix in transaction.message.instructions:
program = transaction.message.account_keys[ix.program_id_index]
if str(program) != str(PUMP_PROGRAM_ID):
continue
ix_data = bytes(ix.data)
if len(ix_data) < 8:
continue
discriminator = struct.unpack("<Q", ix_data[:8])[0]
if discriminator not in (CREATE_DISCRIMINATOR, CREATE_V2_DISCRIMINATOR):
continue
is_v2 = discriminator == CREATE_V2_DISCRIMINATOR
wanted = "create_v2" if is_v2 else "create"
ix_def = next(
(instr for instr in idl["instructions"] if instr["name"] == wanted),
next(instr for instr in idl["instructions"] if instr["name"] == "create"),
)
account_keys = get_account_keys(transaction, ix, loaded_addresses)
if account_keys is None:
print("⚠️ Skipping transaction due to unresolved accounts")
continue
decode = decode_create_v2_instruction if is_v2 else decode_create_instruction
print("\n🔍 Found a creation by decoding the envelope (no logs in this block)")
print_token_info(decode(ix_data, ix_def, account_keys))
return True
return False
def handle_transaction(tx, idl):
def handle_transaction(tx):
"""Detect and print a coin creation in one transaction from a block.
Detection routes on `meta.logMessages`, which the RPC has already decoded
and which reads the same whatever version the transaction is. The envelope
is only opened afterwards, to report address lookup table use, and only when
the installed solders can read it. Gating detection on that decode makes the
example blind to every block carrying a version solders does not handle.
Detection routes on `meta.logMessages`. The RPC has already decoded the
envelope by the time it emits the logs, and they read the same for every
transaction version, so the logs are the one route: a `create` reached by
CPI appears in them and is absent from the envelope's top-level
instructions. Opening the envelope is a separate exercise, done by
`cookbook/pumpfun/decode/pumpfun_decode_transaction_blocksubscribe.py`.
Args:
tx: One entry from a blockSubscribe notification's `transactions`
idl: The parsed pump.fun IDL, kept for the envelope path
"""
meta = tx.get("meta") or {}
logs = meta.get("logMessages")
version = tx.get("version", "legacy")
if logs:
event = find_create_event(logs)
if not event:
return
label = "legacy" if version == "legacy" else f"v{version}"
print(f"\n🔍 Found CreateEvent in a {label} transaction")
print_token_info(event)
# Some providers return blocks without logMessages. The envelope is then
# the only route, and it only works for a version solders can read.
elif not decode_from_envelope(tx, idl):
# The runtime keeps the logs a transaction emitted before it failed, so a
# create that succeeded inside a transaction a later instruction reverted
# still leaves a CreateEvent here. The mint does not exist, and a snipe on
# it fails with InvalidMint, so `err` has to be read to tell the two apart.
if meta.get("err") is not None:
return
logs = meta.get("logMessages")
if not logs:
return
event = find_create_event(logs)
if not event:
return
version = tx.get("version", "legacy")
label = "legacy" if version == "legacy" else f"v{version}"
print(f"\n🔍 Found CreateEvent in a {label} transaction")
print_token_info(event)
loaded_addresses = meta.get("loadedAddresses")
if loaded_addresses:
writable_count = len(loaded_addresses.get("writable", []))
@@ -477,13 +241,10 @@ async def listen_and_decode_create():
"""Main listener function that subscribes to Solana blocks and decodes Pump.fun token creations.
This function:
1. Loads the Pump.fun IDL for instruction parsing
2. Subscribes to blocks mentioning the Pump.fun program
3. Decodes transactions to extract Create/CreateV2 instructions
4. Handles Address Lookup Tables (ALTs) for account resolution
1. Subscribes to blocks mentioning the Pump.fun program
2. Reads each transaction's `meta.logMessages` for a CreateEvent
3. Reports address lookup table use alongside each creation
"""
idl = load_idl("idl/pump_fun_idl.json")
async with websockets.connect(
WSS_ENDPOINT, max_size=WEBSOCKET_MAX_MESSAGE_BYTES
) as websocket:
@@ -524,7 +285,7 @@ async def listen_and_decode_create():
for tx in block["transactions"]:
if not isinstance(tx, dict):
continue
handle_transaction(tx, idl)
handle_transaction(tx)
elif "result" in data:
print("Subscription confirmed")
else:
cookbook/pumpfun/listen/pumpfun_listen_tokens_deshred.py
+21 -15
View File
@@ -41,7 +41,7 @@ Geyser gRPC reference:
https://docs.triton.one/project-yellowstone/dragons-mouth-grpc-subscriptions
Authentication: Basic or X-Token, via GEYSER_ENDPOINT, GEYSER_API_TOKEN and
AUTH_TYPE.
GEYSER_AUTH_TYPE.
"""
import asyncio
@@ -65,8 +65,9 @@ load_dotenv()
GEYSER_ENDPOINT = os.getenv("GEYSER_ENDPOINT")
GEYSER_API_TOKEN = os.getenv("GEYSER_API_TOKEN")
# Authentication type: "x-token" or "basic"
AUTH_TYPE = "x-token"
AUTH_TYPE = os.getenv("GEYSER_AUTH_TYPE", "x-token").lower()
BAD_AUTH_TYPE_MSG = "GEYSER_AUTH_TYPE must be 'x-token' or 'basic'"
PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
@@ -98,7 +99,7 @@ def print_token_info(token_data: dict, signature: str, slot: int) -> None:
# From the instruction args, not the curve: may differ from BondingCurve.creator.
print(f"Creator (args): {token_data.get('creator', 'N/A')}")
print(f"Token Standard: {token_data.get('token_standard', 'N/A')}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', False)}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', 'N/A')}")
print(f"URI: {token_data.get('uri', 'N/A')}")
print(f"Tx version: {token_data.get('tx_version', 'N/A')}")
print(f"Slot: {slot}")
@@ -112,17 +113,22 @@ async def create_geyser_connection() -> geyser_pb2_grpc.GeyserStub:
Returns:
A Geyser stub bound to an authenticated channel
Raises:
ValueError: If GEYSER_AUTH_TYPE names a scheme the endpoint does not take
"""
if AUTH_TYPE == "x-token":
auth = grpc.metadata_call_credentials(
lambda _, callback: callback((("x-token", GEYSER_API_TOKEN),), None)
)
else: # Default to basic auth
elif AUTH_TYPE == "basic":
auth = grpc.metadata_call_credentials(
lambda _, callback: callback(
(("authorization", f"Basic {GEYSER_API_TOKEN}"),), None
)
)
else:
raise ValueError(BAD_AUTH_TYPE_MSG)
creds = grpc.composite_channel_credentials(grpc.ssl_channel_credentials(), auth)
channel = grpc.aio.secure_channel(GEYSER_ENDPOINT, creds)
@@ -223,17 +229,9 @@ def decode_create_instruction(
"associated_bonding_curve": get_account_key(3),
"user": get_account_key(7),
"token_standard": "legacy",
"is_mayhem_mode": False,
}
# create_v2 trailing args are positional and may be truncated on the wire:
# read defensively and report a missing one as unset rather than raising.
is_mayhem_mode = False
if offset < len(ix_data):
is_mayhem_mode = bool(ix_data[offset])
offset += 1
return {
token_info = {
"name": name,
"symbol": symbol,
"uri": uri,
@@ -243,9 +241,17 @@ def decode_create_instruction(
"associated_bonding_curve": get_account_key(3),
"user": get_account_key(5),
"token_standard": "token2022",
"is_mayhem_mode": is_mayhem_mode,
}
# create_v2 trailing args are positional and may be truncated on the wire.
# A missing one is left out of the dict: the caller prints "N/A" for it,
# rather than a False the instruction never carried.
if offset < len(ix_data):
token_info["is_mayhem_mode"] = bool(ix_data[offset])
offset += 1
return token_info
def describe_version(message: geyser_pb2.Message) -> str:
"""Name the transaction format a message is in.
cookbook/pumpfun/listen/pumpfun_listen_tokens_geyser.py
+22 -18
View File
@@ -12,7 +12,7 @@ Geyser gRPC Reference:
https://docs.triton.one/rpc-pool/grpc-subscriptions
Authentication: Supports both Basic and X-Token authentication methods.
Configure via GEYSER_ENDPOINT, GEYSER_API_TOKEN, and AUTH_TYPE variables.
Configure via GEYSER_ENDPOINT, GEYSER_API_TOKEN and GEYSER_AUTH_TYPE.
"""
import asyncio
@@ -36,8 +36,9 @@ load_dotenv()
GEYSER_ENDPOINT = os.getenv("GEYSER_ENDPOINT")
GEYSER_API_TOKEN = os.getenv("GEYSER_API_TOKEN")
# Authentication type: "x-token" or "basic"
AUTH_TYPE = "x-token"
AUTH_TYPE = os.getenv("GEYSER_AUTH_TYPE", "x-token").lower()
BAD_AUTH_TYPE_MSG = "GEYSER_AUTH_TYPE must be 'x-token' or 'basic'"
PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
@@ -74,7 +75,7 @@ def print_token_info(token_data, signature=None, envelope: dict | None = None):
print(f"Creator: {token_data['creator']}")
print(f"Token Standard: {token_data.get('token_standard', 'N/A')}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', False)}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', 'N/A')}")
if "uri" in token_data:
print(f"URI: {token_data['uri']}")
@@ -91,17 +92,23 @@ def print_token_info(token_data, signature=None, envelope: dict | None = None):
async def create_geyser_connection():
"""Establish a secure connection to the Geyser endpoint using the configured auth type."""
"""Establish a secure connection to the Geyser endpoint using the configured auth type.
Raises:
ValueError: If GEYSER_AUTH_TYPE names a scheme the endpoint does not take
"""
if AUTH_TYPE == "x-token":
auth = grpc.metadata_call_credentials(
lambda _, callback: callback((("x-token", GEYSER_API_TOKEN),), None)
)
else: # Default to basic auth
elif AUTH_TYPE == "basic":
auth = grpc.metadata_call_credentials(
lambda _, callback: callback(
(("authorization", f"Basic {GEYSER_API_TOKEN}"),), None
)
)
else:
raise ValueError(BAD_AUTH_TYPE_MSG)
creds = grpc.composite_channel_credentials(grpc.ssl_channel_credentials(), auth)
channel = grpc.aio.secure_channel(GEYSER_ENDPOINT, creds)
@@ -245,7 +252,6 @@ def decode_create_instruction(ix_data: bytes, keys, accounts) -> dict:
"rent": get_account_key(6),
"user": get_account_key(7),
"token_standard": "legacy",
"is_mayhem_mode": False,
}
return token_info
@@ -291,15 +297,6 @@ def decode_create_v2_instruction(ix_data: bytes, keys, accounts) -> dict:
uri = read_string()
creator = read_pubkey()
# CreateV2 trailing args: is_mayhem_mode (bool, 1B), is_cashback_enabled (OptionBool, 1B)
is_mayhem_mode = False
is_cashback_enabled = False
if offset < len(ix_data):
is_mayhem_mode = bool(ix_data[offset])
offset += 1
if offset < len(ix_data):
is_cashback_enabled = bool(ix_data[offset])
token_info = {
"name": name,
"symbol": symbol,
@@ -310,10 +307,17 @@ def decode_create_v2_instruction(ix_data: bytes, keys, accounts) -> dict:
"associated_bonding_curve": get_account_key(3),
"user": get_account_key(5),
"token_standard": "token2022",
"is_mayhem_mode": is_mayhem_mode,
"is_cashback_enabled": is_cashback_enabled,
}
# CreateV2 trailing args: is_mayhem_mode (bool, 1B), is_cashback_enabled
# (OptionBool, 1B). Either may be truncated off the wire; a missing one is
# left out of the dict rather than reported as False.
if offset < len(ix_data):
token_info["is_mayhem_mode"] = bool(ix_data[offset])
offset += 1
if offset < len(ix_data):
token_info["is_cashback_enabled"] = bool(ix_data[offset])
return token_info
cookbook/pumpfun/listen/pumpfun_listen_tokens_logsubscribe.py
+1 -1
View File
@@ -82,7 +82,7 @@ def print_token_info(
print(f"Creator: {token_data['creator']}")
print(f"Token Standard: {token_data.get('token_standard', 'N/A')}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', False)}")
print(f"Mayhem Mode: {token_data.get('is_mayhem_mode', 'N/A')}")
if "uri" in token_data:
print(f"URI: {token_data['uri']}")
cookbook/pumpfun/read/pumpfun_read_quote_mints.py
+3 -1
View File
@@ -172,7 +172,9 @@ async def read_quote_mints(*, stocks_only: bool) -> None:
for mint, reserves, info in sorted(
rows, key=lambda r: r[2].get("symbol") or "~"
):
program = "Token-2022" if info.get("token_2022") else "SPL Token"
program = {True: "Token-2022", False: "SPL Token"}.get(
info.get("token_2022"), "-"
)
paused = {True: "YES", False: "no"}.get(info.get("paused"), "-")
multiplier = info.get("multiplier") or "-"
decimals = info.get("decimals")
cookbook/pumpfun/read/pumpfun_watch_curve_progress.py
+20 -19
View File
@@ -57,8 +57,9 @@ _LEN_WITH_MAYHEM: Final[int] = 74
_LEN_WITH_CASHBACK: Final[int] = 75
_LEN_WITH_QUOTE_MINT: Final[int] = 107
# Only used if the Global account cannot be read: 1B supply less 206.9M reserved.
FALLBACK_INITIAL_REAL_TOKEN_RESERVES: Final[float] = 793_100_000.0
_NO_BASELINE_MSG: Final[str] = (
"Cannot read initial_real_token_reserves from the pump.fun Global account"
)
async def read_quote_decimals(conn: AsyncClient, quote_mint: Pubkey) -> int:
@@ -157,17 +158,14 @@ def parse_curve_state(data: bytes) -> dict:
creator_bytes = data[49:81] # 8 (discriminator) + 41 (base fields) = 49
result["creator"] = Pubkey.from_bytes(creator_bytes)
# Parse is_mayhem_mode if present
if data_length >= _LEN_WITH_MAYHEM: # Has mayhem mode field
# Both flags are absent from a curve laid out before they were added. Leave
# them out of the result rather than reporting an unset field as false.
if data_length >= _LEN_WITH_MAYHEM:
result["is_mayhem_mode"] = bool(data[81])
else:
result["is_mayhem_mode"] = False
# Parse is_cashback_coin if present (added in late-Feb 2026 cashback upgrade)
# is_cashback_coin arrived with the late-Feb 2026 cashback upgrade.
if data_length >= _LEN_WITH_CASHBACK:
result["is_cashback_coin"] = bool(data[82])
else:
result["is_cashback_coin"] = False
return result
@@ -183,17 +181,20 @@ async def fetch_initial_real_token_reserves(client: AsyncClient) -> float:
client: Connected RPC client
Returns:
Initial real token reserves in whole tokens, or the fallback constant
Initial real token reserves in whole tokens
Raises:
ValueError: If Global is missing or carries a zero at that offset. The
baseline is the 0% mark every progress figure is measured against,
so a guessed one misreports every coin the run touches.
"""
try:
resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64")
data = resp.value.data
raw = struct.unpack_from("<Q", data, 89)[0]
if raw:
return raw / 10**TOKEN_DECIMALS
except Exception as e: # noqa: BLE001 - fall back rather than abort the poller
print(f"⚠️ Could not read Global, using the fallback baseline: {e}")
return FALLBACK_INITIAL_REAL_TOKEN_RESERVES
resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64")
if resp.value is None or len(resp.value.data) < 89 + 8:
raise ValueError(_NO_BASELINE_MSG)
raw = struct.unpack_from("<Q", resp.value.data, 89)[0]
if not raw:
raise ValueError(_NO_BASELINE_MSG)
return raw / 10**TOKEN_DECIMALS
def print_curve_status(state: dict, baseline: float, quote_unit: int) -> None:
cookbook/pumpfun/trade/pumpfun_instructions_v2.py
+21 -6
View File
@@ -49,6 +49,8 @@ LAMPORTS_PER_SOL = 1_000_000_000
# Account discriminator for the BondingCurve account.
BONDING_CURVE_DISCRIMINATOR = struct.pack("<Q", 6966180631402821399)
_SHORT_CURVE_MSG = "Bonding curve data ends before {name} at byte {offset}"
# Instruction discriminators (first 8 bytes of sha256("global:<name>")).
BUY_V2_DISCRIMINATOR = bytes([184, 23, 238, 97, 103, 197, 211, 61])
SELL_V2_DISCRIMINATOR = bytes([93, 246, 130, 60, 231, 233, 64, 178])
@@ -376,8 +378,12 @@ class BondingCurveState:
self.real_sol_reserves = self.real_quote_reserves
self.creator = self._read_pubkey(data, self._CREATOR_OFFSET)
self.is_mayhem_mode = self._read_flag(data, self._MAYHEM_OFFSET)
self.is_cashback_coin = self._read_flag(data, self._CASHBACK_OFFSET)
self.is_mayhem_mode = self._read_flag(
data, self._MAYHEM_OFFSET, "is_mayhem_mode"
)
self.is_cashback_coin = self._read_flag(
data, self._CASHBACK_OFFSET, "is_cashback_coin"
)
raw_quote_mint = self._read_pubkey(data, self._QUOTE_MINT_OFFSET)
self.quote_mint = normalize_quote_mint(raw_quote_mint)
self.is_sol_paired = is_sol_paired(raw_quote_mint)
@@ -398,17 +404,26 @@ class BondingCurveState:
return Pubkey.from_bytes(data[offset : offset + 32])
@staticmethod
def _read_flag(data: bytes, offset: int) -> bool:
"""Read a single-byte bool if the data extends that far.
def _read_flag(data: bytes, offset: int, name: str) -> bool:
"""Read a single-byte bool, refusing to guess when the field is absent.
Args:
data: Raw account data
offset: Byte offset
name: Field name, used in the error
Returns:
Flag value, or False if the field is absent
Flag value
Raises:
ValueError: If the data ends before the flag. is_mayhem_mode picks
the fee recipient for a buy, and the wrong one reverts with
NotAuthorized (6000), so defaulting it only moves the failure
to somewhere harder to read.
"""
return bool(data[offset]) if len(data) > offset else False
if len(data) <= offset:
raise ValueError(_SHORT_CURVE_MSG.format(name=name, offset=offset))
return bool(data[offset])
def price_per_token(self) -> float:
"""Current price in whole quote units per whole token.
cookbook/pumpfun/trade/pumpfun_snipe_token_geyser.py
+11 -3
View File
@@ -84,7 +84,9 @@ RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
# Geyser endpoints
GEYSER_ENDPOINT = os.environ.get("GEYSER_ENDPOINT")
GEYSER_API_TOKEN = os.environ.get("GEYSER_API_TOKEN")
AUTH_TYPE = os.environ.get("GEYSER_AUTH_TYPE", "x-token") # Default to x-token
AUTH_TYPE = os.environ.get("GEYSER_AUTH_TYPE", "x-token").lower()
BAD_AUTH_TYPE_MSG = "GEYSER_AUTH_TYPE must be 'x-token' or 'basic'"
PUMP_CREATE_DISCRIMINATOR = struct.pack("<Q", 8576854823835016728)
PUMP_CREATE_V2_DISCRIMINATOR = bytes([214, 144, 76, 236, 95, 139, 49, 180])
@@ -145,17 +147,23 @@ def calculate_pump_curve_price(curve_state: pump_v2.BondingCurveState) -> float:
async def create_geyser_connection():
"""Establish a secure connection to the Geyser endpoint using the configured auth type."""
"""Establish a secure connection to the Geyser endpoint using the configured auth type.
Raises:
ValueError: If GEYSER_AUTH_TYPE names a scheme the endpoint does not take
"""
if AUTH_TYPE == "x-token":
auth = grpc.metadata_call_credentials(
lambda _, callback: callback((("x-token", GEYSER_API_TOKEN),), None)
)
else: # Default to basic auth
elif AUTH_TYPE == "basic":
auth = grpc.metadata_call_credentials(
lambda _, callback: callback(
(("authorization", f"Basic {GEYSER_API_TOKEN}"),), None
)
)
else:
raise ValueError(BAD_AUTH_TYPE_MSG)
creds = grpc.composite_channel_credentials(grpc.ssl_channel_credentials(), auth)
channel = grpc.aio.secure_channel(GEYSER_ENDPOINT, creds)
cookbook/pumpswap/pumpswap_buy_token.py
+65 -5
View File
@@ -147,6 +147,28 @@ VOLUME_TRACKING_ENABLED = 1 # 1 = true, 0 = false
# ============================================================================
# coin_creator sits after discriminator(8) + pool_bump(1) + index(2) + six
# pubkeys(192) + lp_supply(8). It is set only on a pool the coin graduated into
# and left at the default on every other pool, so it is what tells the canonical
# pool from a copy anyone can open against the same mint.
POOL_COIN_CREATOR_OFFSET = 211
def read_pool_coin_creator(data: bytes) -> Pubkey:
"""Read a pool account's coin_creator.
Args:
data: Raw pool account data, discriminator included
Returns:
The pubkey, or the default one if the account is too short to hold it
"""
end = POOL_COIN_CREATOR_OFFSET + 32
if len(data) < end:
return DEFAULT_COIN_CREATOR
return Pubkey.from_bytes(data[POOL_COIN_CREATOR_OFFSET:end])
async def get_market_address_by_base_mint(
client: AsyncClient, base_mint_address: Pubkey, amm_program_id: Pubkey
) -> Pubkey:
@@ -161,14 +183,35 @@ async def get_market_address_by_base_mint(
amm_program_id: PUMP AMM program address
Returns:
Address of the AMM pool (market) for the token
Address of the coin's canonical pool
Raises:
ValueError: If the coin has no canonical pool, or carries more than one
"""
# MemcmpOpts takes the bytes base58-encoded, which a Pubkey's str already is.
filters = [MemcmpOpts(offset=POOL_BASE_MINT_OFFSET, bytes=str(base_mint_address))]
response = await client.get_program_accounts(
amm_program_id, encoding="base64", filters=filters
)
return response.value[0].pubkey
canonical = [
account.pubkey
for account in response.value
if read_pool_coin_creator(bytes(account.account.data)) != DEFAULT_COIN_CREATOR
]
if not canonical:
raise ValueError(
f"No canonical PumpSwap pool for {base_mint_address}. "
f"{len(response.value)} pool(s) carry this base mint, none of them "
f"created by graduation. Pass --pool to trade a specific one."
)
if len(canonical) > 1:
listed = ", ".join(str(pool) for pool in canonical)
raise ValueError(
f"{len(canonical)} canonical pools for {base_mint_address}: {listed}. "
f"Pass --pool to choose one."
)
return canonical[0]
async def get_market_data(client: AsyncClient, market_address: Pubkey) -> dict:
@@ -755,17 +798,20 @@ async def buy_pump_swap(
# ============================================================================
async def buy(token_mint: Pubkey, sol_amount: float, slippage: float) -> None:
async def buy(
token_mint: Pubkey, sol_amount: float, slippage: float, pool: Pubkey | None = None
) -> None:
"""Execute the complete buy flow.
Args:
token_mint: The coin to buy
sol_amount: SOL to spend
slippage: Maximum acceptable price movement
pool: Trade this pool rather than the coin's canonical one
"""
async with AsyncClient(RPC_ENDPOINT, timeout=120) as client:
# Step 1: Find the pool address for our token
market_address = await get_market_address_by_base_mint(
market_address = pool or await get_market_address_by_base_mint(
client, token_mint, PUMP_AMM_PROGRAM_ID
)
@@ -829,10 +875,24 @@ def main() -> None:
default=DEFAULT_SLIPPAGE,
help=f"Maximum acceptable price movement (default {DEFAULT_SLIPPAGE})",
)
parser.add_argument(
"--pool",
help=(
"Trade this pool instead of the coin's canonical one. Needed for a "
"pool nobody graduated into, which the default selection refuses."
),
)
args = parser.parse_args()
try:
asyncio.run(buy(Pubkey.from_string(args.mint), args.amount, args.slippage))
asyncio.run(
buy(
Pubkey.from_string(args.mint),
args.amount,
args.slippage,
Pubkey.from_string(args.pool) if args.pool else None,
)
)
except ValueError as e:
print(e)
sys.exit(1)
cookbook/pumpswap/pumpswap_find_pool.py
+71 -12
View File
@@ -22,10 +22,45 @@ load_dotenv()
RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
PUMP_AMM_PROGRAM_ID = Pubkey.from_string("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA")
# coin_creator sits after discriminator(8) + pool_bump(1) + index(2) + six
# pubkeys(192) + lp_supply(8). It is set only on a pool the coin graduated into
# and left at the default on every other pool, so it is what tells the canonical
# pool from a copy anyone can open against the same mint.
POOL_COIN_CREATOR_OFFSET = 211
DEFAULT_COIN_CREATOR = Pubkey.default()
async def get_market_address_by_base_mint(
def read_pool_coin_creator(data: bytes) -> Pubkey:
"""Read a pool account's coin_creator.
Args:
data: Raw pool account data, discriminator included
Returns:
The pubkey, or the default one if the account is too short to hold it
"""
end = POOL_COIN_CREATOR_OFFSET + 32
if len(data) < end:
return DEFAULT_COIN_CREATOR
return Pubkey.from_bytes(data[POOL_COIN_CREATOR_OFFSET:end])
async def get_pools_by_base_mint(
base_mint_address: Pubkey, amm_program_id: Pubkey
):
) -> tuple[list[Pubkey], list[Pubkey]]:
"""Find every pool holding this base mint, split by whether it is canonical.
Anyone can open a pool against any mint, and getProgramAccounts returns them
in no defined order, so the first match is an arbitrary pool rather than the
coin's own.
Args:
base_mint_address: The coin's mint
amm_program_id: PUMP AMM program address
Returns:
Canonical pools and the rest, each in the order the RPC returned them
"""
async with AsyncClient(RPC_ENDPOINT, timeout=120) as client:
# Define the offset for base_mint field
offset = 43
@@ -41,12 +76,16 @@ async def get_market_address_by_base_mint(
filters=filters,
)
pool_addresses = [account.pubkey for account in response.value]
if not pool_addresses:
# No pool means the coin has not graduated off its bonding curve
# yet, which is the common case rather than an error.
return None
return pool_addresses[0]
canonical, others = [], []
for account in response.value:
target = (
canonical
if read_pool_coin_creator(bytes(account.account.data))
!= DEFAULT_COIN_CREATOR
else others
)
target.append(account.pubkey)
return canonical, others
# Same offset in SPL Token and Token-2022: extensions are appended after it.
@@ -142,22 +181,42 @@ async def show_pool(token_mint: Pubkey) -> None:
Args:
token_mint: The coin whose pool to look up
"""
market_address = await get_market_address_by_base_mint(
token_mint, PUMP_AMM_PROGRAM_ID
)
if market_address is None:
canonical, others = await get_pools_by_base_mint(token_mint, PUMP_AMM_PROGRAM_ID)
if not canonical and not others:
print(
f"No PumpSwap pool for {token_mint}.\n"
"It has not graduated off its bonding curve yet — trade it with\n"
"cookbook/pumpfun/trade/pumpfun_buy_token_v2.py instead."
)
return
if not canonical:
print(
f"No canonical pool for {token_mint}. The {len(others)} pool(s) below "
"carry its base mint but none was created by graduation:"
)
for pool in others:
print(f" {pool}")
return
market_address = canonical[0]
print(market_address)
market_data = await get_market_data(market_address)
for key, value in market_data.items():
print(f" {key}: {value}")
if len(canonical) > 1:
print(f"\nWARNING: {len(canonical)} canonical pools carry this base mint:")
for pool in canonical[1:]:
print(f" {pool}")
if others:
print(
f"\n{len(others)} further pool(s) carry this base mint and are not "
"canonical. Trading one means passing --pool to the buy/sell scripts:"
)
for pool in others:
print(f" {pool}")
# Quote against effective reserves. Upstream's release note says
# virtual_quote_reserves is 0 on all pools; that is out of date — pools carry
# 17.5845 SOL of virtual reserves, so quoting off the raw vault balance
cookbook/pumpswap/pumpswap_sell_token.py
+62 -5
View File
@@ -133,6 +133,28 @@ COMPUTE_UNIT_BUDGET = 150_000 # Max compute units for transaction
# ============================================================================
# coin_creator sits after discriminator(8) + pool_bump(1) + index(2) + six
# pubkeys(192) + lp_supply(8). It is set only on a pool the coin graduated into
# and left at the default on every other pool, so it is what tells the canonical
# pool from a copy anyone can open against the same mint.
POOL_COIN_CREATOR_OFFSET = 211
def read_pool_coin_creator(data: bytes) -> Pubkey:
"""Read a pool account's coin_creator.
Args:
data: Raw pool account data, discriminator included
Returns:
The pubkey, or the default one if the account is too short to hold it
"""
end = POOL_COIN_CREATOR_OFFSET + 32
if len(data) < end:
return DEFAULT_COIN_CREATOR
return Pubkey.from_bytes(data[POOL_COIN_CREATOR_OFFSET:end])
async def get_market_address_by_base_mint(
client: AsyncClient, base_mint_address: Pubkey, amm_program_id: Pubkey
) -> Pubkey:
@@ -147,14 +169,35 @@ async def get_market_address_by_base_mint(
amm_program_id: PUMP AMM program address
Returns:
Address of the AMM pool (market) for the token
Address of the coin's canonical pool
Raises:
ValueError: If the coin has no canonical pool, or carries more than one
"""
# MemcmpOpts takes the bytes base58-encoded, which a Pubkey's str already is.
filters = [MemcmpOpts(offset=POOL_BASE_MINT_OFFSET, bytes=str(base_mint_address))]
response = await client.get_program_accounts(
amm_program_id, encoding="base64", filters=filters
)
return response.value[0].pubkey
canonical = [
account.pubkey
for account in response.value
if read_pool_coin_creator(bytes(account.account.data)) != DEFAULT_COIN_CREATOR
]
if not canonical:
raise ValueError(
f"No canonical PumpSwap pool for {base_mint_address}. "
f"{len(response.value)} pool(s) carry this base mint, none of them "
f"created by graduation. Pass --pool to trade a specific one."
)
if len(canonical) > 1:
listed = ", ".join(str(pool) for pool in canonical)
raise ValueError(
f"{len(canonical)} canonical pools for {base_mint_address}: {listed}. "
f"Pass --pool to choose one."
)
return canonical[0]
async def get_market_data(client: AsyncClient, market_address: Pubkey) -> dict:
@@ -713,16 +756,17 @@ async def sell_pump_swap(
# ============================================================================
async def sell(token_mint: Pubkey, slippage: float) -> None:
async def sell(token_mint: Pubkey, slippage: float, pool: Pubkey | None = None) -> None:
"""Execute the complete sell flow.
Args:
token_mint: The coin to sell
slippage: Maximum acceptable price movement
pool: Trade this pool rather than the coin's canonical one
"""
async with AsyncClient(RPC_ENDPOINT, timeout=120) as client:
# Step 1: Find the pool address for our token
market_address = await get_market_address_by_base_mint(
market_address = pool or await get_market_address_by_base_mint(
client, token_mint, PUMP_AMM_PROGRAM_ID
)
@@ -780,10 +824,23 @@ def main() -> None:
default=DEFAULT_SLIPPAGE,
help=f"Maximum acceptable price movement (default {DEFAULT_SLIPPAGE})",
)
parser.add_argument(
"--pool",
help=(
"Trade this pool instead of the coin's canonical one. Needed for a "
"pool nobody graduated into, which the default selection refuses."
),
)
args = parser.parse_args()
try:
asyncio.run(sell(Pubkey.from_string(args.mint), args.slippage))
asyncio.run(
sell(
Pubkey.from_string(args.mint),
args.slippage,
Pubkey.from_string(args.pool) if args.pool else None,
)
)
except ValueError as e:
print(e)
sys.exit(1)
tools/compare_listeners.py
+19 -13
View File
@@ -63,8 +63,9 @@ PUMPPORTAL_WS_URL = "wss://pumpportal.fun/api/data"
# Test duration in seconds
TEST_DURATION = 30
# Geyser authentication type: "x-token" or "basic"
GEYSER_AUTH_TYPE = "x-token"
GEYSER_AUTH_TYPE = os.getenv("GEYSER_AUTH_TYPE", "x-token").lower()
BAD_AUTH_TYPE_MSG = "GEYSER_AUTH_TYPE must be 'x-token' or 'basic'"
class DetectionTracker:
@@ -696,21 +697,26 @@ async def listen_geyser_grpc(
if known_tokens is None:
known_tokens = set()
# Built once, outside the retry loop: the reconnect handler below catches
# every exception and sleeps, so a bad auth type raised in there would
# reconnect forever instead of reporting a misconfiguration.
if GEYSER_AUTH_TYPE == "x-token":
auth = grpc.metadata_call_credentials(
lambda _context, callback: callback((("x-token", api_token),), None)
)
elif GEYSER_AUTH_TYPE == "basic":
auth = grpc.metadata_call_credentials(
lambda _context, callback: callback(
(("authorization", f"Basic {api_token}"),), None
)
)
else:
raise ValueError(BAD_AUTH_TYPE_MSG)
while True:
try:
print(f"[INFO] Connecting Geyser gRPC listener to {provider_name}...")
if GEYSER_AUTH_TYPE == "x-token":
auth = grpc.metadata_call_credentials(
lambda _context, callback: callback((("x-token", api_token),), None)
)
else:
auth = grpc.metadata_call_credentials(
lambda _context, callback: callback(
(("authorization", f"Basic {api_token}"),), None
)
)
creds = grpc.composite_channel_credentials(
grpc.ssl_channel_credentials(), auth
)
tools/compare_migration_listeners.py
+6 -10
View File
@@ -432,14 +432,6 @@ def parse_migrate_instruction(data):
return None
def is_transaction_successful(logs):
"""Check if a transaction was successful based on log messages"""
for log in logs:
if "AnchorError thrown" in log or "Error" in log:
return False
return True
# ============ WEBSOCKET LISTENERS ============
@@ -492,8 +484,12 @@ async def listen_for_migrations(wss_url, provider_name, tracker, known_events=No
log_data = data["params"]["result"]["value"]
logs = log_data.get("logs", [])
# Skip failed transactions
if not is_transaction_successful(logs):
# Skip failed transactions. The notification carries
# `err` alongside the logs; scanning the log text for
# "Error" instead drops any migration whose logs merely
# contain the word, in the tool that exists to measure
# which transport saw what.
if log_data.get("err") is not None:
continue
# Skip if not a Migrate instruction
tools/live_listener_matrix.py
+22 -5
View File
@@ -18,6 +18,7 @@ Usage:
"""
import asyncio
import contextlib
import os
import sys
from dataclasses import dataclass, field
@@ -54,7 +55,11 @@ BUY_SLIPPAGE = 0.3
DETECT_TIMEOUT_SECONDS = 150.0
# Time to let a close/sell finalize before reading balances or account state.
SETTLE_SECONDS = 20
ALL_LISTENERS = ("geyser", "logs", "blocks", "pumpportal")
# Taken from the factory so a listener added there cannot be silently skipped
# here while the run still reports every listener passing.
ALL_LISTENERS = tuple(
ListenerFactory.get_platform_compatible_listeners(Platform.PUMP_FUN)
)
@dataclass
@@ -103,7 +108,11 @@ async def detect(listener_type: str) -> TokenInfo | None:
"""Wait for one new pump.fun token from the given listener.
Returns:
TokenInfo, or None on timeout
TokenInfo, or None if nothing was created before the timeout
Raises:
Exception: Whatever the listener raised, rather than reporting it as
an idle market
"""
listener = make_listener(listener_type)
seen: list[TokenInfo] = []
@@ -114,12 +123,20 @@ async def detect(listener_type: str) -> TokenInfo | None:
task = asyncio.create_task(listener.listen_for_tokens(on_token))
try:
for _ in range(int(DETECT_TIMEOUT_SECONDS / 0.5)):
if seen:
if seen or task.done():
break
await asyncio.sleep(0.5)
finally:
task.cancel()
await asyncio.sleep(0)
if not task.done():
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
# Bad credentials, a dead endpoint and a quiet market all leave `seen`
# empty. Re-raise what the listener hit, so the harness that exists to
# certify listeners cannot report a broken one as an idle one.
if task.done() and not task.cancelled():
task.result()
return seen[0] if seen else None
tools/simulate_bot_buy_path.py
+94 -49
View File
@@ -8,7 +8,8 @@ before submission and simulates it instead. This exercises the listener ->
event parser -> curve manager -> address provider -> instruction builder
chain as a unit.
No funds move: `build_and_send_transaction` is monkeypatched to simulate.
No funds move: the RPC client's `send_transaction` is monkeypatched to
simulate whatever the bot handed it.
Usage:
uv run tools/simulate_bot_buy_path.py
@@ -16,21 +17,20 @@ Usage:
"""
import asyncio
import contextlib
import os
import struct
import sys
from base64 import b64encode
from pathlib import Path
from types import SimpleNamespace
PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT / "src"))
from dotenv import load_dotenv # noqa: E402
from solders.compute_budget import ( # noqa: E402
set_compute_unit_limit,
set_compute_unit_price,
)
from solders.message import MessageV0 # noqa: E402
from solders.transaction import VersionedTransaction # noqa: E402
from solders.pubkey import Pubkey # noqa: E402
from solders.signature import Signature # noqa: E402
from core.client import SolanaClient # noqa: E402
from core.priority_fee.manager import PriorityFeeManager # noqa: E402
@@ -79,17 +79,69 @@ async def wait_for_token(timeout_seconds: float = 90.0) -> TokenInfo | None:
task = asyncio.create_task(listener.listen_for_tokens(on_token))
try:
for _ in range(int(timeout_seconds / 0.5)):
if seen:
if seen or task.done():
break
await asyncio.sleep(0.5)
finally:
task.cancel()
if not task.done():
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
# Bad credentials, a dead endpoint and a quiet market all leave `seen`
# empty. Re-raise what the listener hit so they stay distinguishable.
if task.done() and not task.cancelled():
task.result()
return seen[0] if seen else None
def install_simulation_hook(client: SolanaClient) -> dict:
"""Replace transaction submission with simulation.
COMPUTE_BUDGET_PROGRAM = Pubkey.from_string(
"ComputeBudget111111111111111111111111111111"
)
# Compute Budget instruction tags, from the program's own enum.
_SET_CU_LIMIT = 2
_SET_CU_PRICE = 3
_SET_DATA_SIZE_LIMIT = 4
def read_compute_budget(message) -> dict:
"""Recover the compute budget settings from a compiled transaction.
Args:
message: The compiled `MessageV0` about to be submitted
Returns:
cu_limit, priority_fee and data_size_limit, each None if the
transaction carries no instruction setting it
"""
found: dict = {"cu_limit": None, "priority_fee": None, "data_size_limit": None}
for ix in message.instructions:
try:
program = message.account_keys[ix.program_id_index]
except IndexError:
continue
if program != COMPUTE_BUDGET_PROGRAM:
continue
data = bytes(ix.data)
if not data:
continue
if data[0] == _SET_CU_LIMIT and len(data) >= 5:
found["cu_limit"] = struct.unpack_from("<I", data, 1)[0]
elif data[0] == _SET_CU_PRICE and len(data) >= 9:
found["priority_fee"] = struct.unpack_from("<Q", data, 1)[0]
elif data[0] == _SET_DATA_SIZE_LIMIT and len(data) >= 5:
found["data_size_limit"] = struct.unpack_from("<I", data, 1)[0]
return found
async def install_simulation_hook(client: SolanaClient) -> dict:
"""Simulate the transaction the bot built, instead of sending it.
Hooks the RPC client's `send_transaction`, so the compute budget preamble,
the blockhash and the message compilation are all the bot's own work. A
hook on `build_and_send_transaction` has to rebuild those, and a rebuild
that misses an instruction simulates something the bot would never send.
Args:
client: Client whose send path should be intercepted
@@ -98,28 +150,9 @@ def install_simulation_hook(client: SolanaClient) -> dict:
Dict that will be populated with the simulation outcome
"""
outcome: dict = {}
rpc = await client.get_client()
async def simulate_instead(
instructions,
signer_keypair,
skip_preflight=True,
max_retries=3,
priority_fee=None,
compute_unit_limit=None,
account_data_size_limit=None,
):
preamble = []
if compute_unit_limit:
preamble.append(set_compute_unit_limit(compute_unit_limit))
if priority_fee:
preamble.append(set_compute_unit_price(priority_fee))
blockhash = await client.get_latest_blockhash()
message = MessageV0.try_compile(
signer_keypair.pubkey(), [*preamble, *instructions], [], blockhash
)
transaction = VersionedTransaction(message, [signer_keypair])
async def simulate_instead(transaction, *_args, **_kwargs):
response = await client.post_rpc(
{
"jsonrpc": "2.0",
@@ -137,24 +170,23 @@ def install_simulation_hook(client: SolanaClient) -> dict:
}
)
value = (response or {}).get("result", {}).get("value", {})
message = transaction.message
outcome.update(
{
"err": value.get("err"),
"units": value.get("unitsConsumed"),
"logs": value.get("logs") or [],
"cu_limit": compute_unit_limit,
"priority_fee": priority_fee,
"instruction_count": len(instructions),
"account_count": len(instructions[-1].accounts),
"instruction_count": len(message.instructions),
"account_count": len(message.instructions[-1].accounts),
**read_compute_budget(message),
}
)
# Returning a sentinel signature: confirm_transaction is stubbed below.
return "SIMULATED"
return SimpleNamespace(value=Signature.default())
async def never_confirm(_signature, **_kwargs):
return False
client.build_and_send_transaction = simulate_instead
rpc.send_transaction = simulate_instead
client.confirm_transaction = never_confirm
return outcome
@@ -167,10 +199,21 @@ async def main() -> int:
"""
extreme_fast = "--no-extreme-fast" not in sys.argv
# Built before detection, as the bot builds it at startup: the client's
# blockhash updater needs a cycle to land, and build_and_send_transaction
# reads the cached value rather than fetching one.
client = SolanaClient(os.environ["SOLANA_NODE_RPC_ENDPOINT"])
wallet = Wallet(os.environ["SOLANA_PRIVATE_KEY"])
print("Waiting for a fresh pump.fun coin via the bot's geyser listener...")
token_info = await wait_for_token()
try:
token_info = await wait_for_token()
except BaseException:
await client.close()
raise
if token_info is None:
print("No coin detected before timeout.")
await client.close()
return 2
print(f"\ndetected: {token_info.symbol} ({token_info.mint})")
@@ -180,8 +223,6 @@ async def main() -> int:
print(f"state_from_event={token_info.state_from_event} (True = zero-RPC buy path)")
print(f"extreme_fast_mode={extreme_fast}\n")
client = SolanaClient(os.environ["SOLANA_NODE_RPC_ENDPOINT"])
wallet = Wallet(os.environ["SOLANA_PRIVATE_KEY"])
priority_fee_manager = PriorityFeeManager(
client=client,
enable_dynamic_fee=False,
@@ -191,7 +232,7 @@ async def main() -> int:
hard_cap=1_000_000,
)
outcome = install_simulation_hook(client)
outcome = await install_simulation_hook(client)
buyer = PlatformAwareBuyer(
client,
wallet,
@@ -221,11 +262,13 @@ async def main() -> int:
return 1
print("simulated buy:")
print(f" instructions: {outcome['instruction_count']}")
print(f" instructions: {outcome['instruction_count']}")
print(f" trade accounts: {outcome['account_count']}")
print(f" cu_limit: {outcome['cu_limit']}")
print(f" unitsConsumed: {outcome['units']}")
print(f" err: {outcome['err']}")
print(f" cu_limit: {outcome['cu_limit']}")
print(f" data size cap: {outcome['data_size_limit']}")
print(f" priority fee: {outcome['priority_fee']}")
print(f" unitsConsumed: {outcome['units']}")
print(f" err: {outcome['err']}")
if outcome["err"]:
for line in outcome["logs"]:
@@ -233,8 +276,10 @@ async def main() -> int:
print(f" {line}")
return 1
headroom = outcome["cu_limit"] - (outcome["units"] or 0)
print(f"\nCU headroom: {headroom} ({headroom / outcome['cu_limit']:.0%})")
cu_limit = outcome["cu_limit"]
if cu_limit:
headroom = cu_limit - (outcome["units"] or 0)
print(f"\nCU headroom: {headroom} ({headroom / cu_limit:.0%})")
print("Buy path validated end to end against live mainnet state.")
return 0