From 678fa19fe477fbbfcddf6305343eba59ae79231e Mon Sep 17 00:00:00 2001 From: Anton Sauchyk <43762166+smypmsa@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:52:04 +0200 Subject: [PATCH] fix(learning-examples): replace getProgramAccounts scan in get_graduating_tokens (#182) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(learning-examples): replace getProgramAccounts scan in get_graduating_tokens The pump.fun program owns over 10M accounts and no provider will scan it, so get_graduating_tokens.py could not run at all (#178). getProgramAccountsV2 is not a fix: it is a provider extension rather than core Agave, and its `limit` is a scan budget, not a result count, so one filtered answer costs ~1000 sequential pages. Rewrite discovery onto filtered programSubscribe, which applies dataSize and memcmp server-side and is accepted even by public api.mainnet-beta.solana.com. Every write to a curve pushes the full 151-byte account, so progress is computed per update with no accumulated state. Add a Geyser sibling that reports the same thing with the slot and signature behind each update. Also fix two bugs that would have survived the rewrite: the mint lookup queried SPL Token, which returns nothing for the Token-2022 ATAs that every create_v2 coin uses, and the threshold was a hardcoded constant rather than Global.initial_real_token_reserves. Closes #178 Co-Authored-By: Claude Opus 5 (1M context) * fix(learning-examples): drop the graduation cutoff that filtered nothing zero_prefix_gate offered a cutoff so high that no coin could fail it, so for any --min-progress below 64.5% the subscription was unfiltered while the banner reported a filter as active. Offer only the three cutoffs that actually narrow, and say plainly when none applies. Rewrite the threshold notes in both scripts in plain English: which cutoffs exist, whether a given threshold gets one, and the part that matters — the pre-filter saves bandwidth but does not decide the answer, so the requested percentage is honoured either way. The banner now names the cutoff as a percentage instead of byte offsets. Both directions were checked against mainnet by running the filtered and unfiltered subscriptions side by side for a minute, on both transports: the filtered stream matched the below-cutoff set exactly, with 143 of 168 curves above the cutoff on WebSocket and 128 of 154 on Geyser. Also document the two scripts in the README example table, and record under throughput that getProgramAccounts over the whole pump.fun program is no longer served by any provider. Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- CLAUDE.md | 21 +- README.md | 4 +- .../get_graduating_tokens.py | 544 ++++++++++++++---- .../get_graduating_tokens_geyser.py | 497 ++++++++++++++++ 4 files changed, 945 insertions(+), 121 deletions(-) create mode 100644 learning-examples/bonding-curve-progress/get_graduating_tokens_geyser.py diff --git a/CLAUDE.md b/CLAUDE.md index c12d1aa..5028859 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -182,8 +182,25 @@ offline and only visible after a couple of minutes against mainnet. - **`getProgramAccounts` over the whole pump program is rejected** by current providers: *"Too many accounts requested (10000001 pubkeys) … use getProgramAccountsV2 with pagination"*. It still works against pump-amm, which - is small enough. `bonding-curve-progress/get_graduating_tokens.py` is knowingly - broken on this and needs the V2 pagination rewrite. + is small enough. Don't take that error message as a fix: `getProgramAccountsV2` + is a provider extension (Helius, Solana Tracker), **not core Agave**, and its + `limit` is a *scan* budget rather than a result count — a page can legally + return zero accounts and a non-null `paginationKey`, so one filtered answer over + the pump program costs ~1000 sequential pages. Reach for a filtered + subscription instead; see the two `get_graduating_tokens*.py` examples. +- **Filtered `programSubscribe` on the pump program is the portable way to find + curves by state.** `dataSize` + `memcmp` are applied server-side, and it is + accepted even by the public `api.mainnet-beta.solana.com`. `memcmp` only matches + exact bytes, so it cannot express "reserves below X" — only a handful of fixed + cutoffs. Treat it as a bandwidth saver and do the real comparison client-side; + don't assume a threshold is being enforced upstream. Geyser's account filters + have the same shape and add the slot and signature. +- **Resolve a curve's mint under Token-2022, not SPL Token.** The curve account + has no mint field and `["bonding-curve", mint]` is not reversible, so the mint + comes from the associated bonding curve ATA — which is Token-2022 for every + `create_v2` coin. `get_token_accounts_by_owner` with the SPL Token program + returns an empty list for all of them, silently. Verified four for four on live + curves, each confirmed by re-deriving the curve PDA from the recovered mint. - **`SetLoadedAccountsDataSizeLimit` must stay generous: 16 MB, not 512 KB.** Verified by simulation on a Token-2022 mint with extensions — 512 KB and 4 MB both fail `MaxLoadedAccountsDataSizeExceeded` with `unitsConsumed=0` (never diff --git a/README.md b/README.md index 923402e..4449461 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,7 @@ Standalone scripts, runnable with `uv run `. No bot config needed — they |---|---| | `listen-new-tokens/` | One listener per method (`logs`, `blocks`, `geyser`, `pumpportal`) plus `compare_listeners.py` to race them | | `listen-migrations/` | Detect a token graduating from the bonding curve to PumpSwap, via the migration wrapper program or new pool accounts | -| `bonding-curve-progress/` | Curve state, progress polling, and tokens close to graduating | +| `bonding-curve-progress/` | Curve state, progress polling, and a live watch for coins close to graduating — over WebSocket (`get_graduating_tokens.py`) or Geyser (`get_graduating_tokens_geyser.py`), both taking `--min-progress` | | `pumpswap/` | Manual buy/sell against the PumpSwap AMM, and pool discovery | | `letsbonk-buy-sell/` | Manual exact-in / exact-out buys and sells on letsbonk.fun | | `copy-trading/` | Watch another wallet's transactions | @@ -149,6 +149,8 @@ Related docs: [Listening to pump.fun migrations](https://docs.chainstack.com/doc Every node provider has its own limits — method availability, requests per second, plan-specific caps. Consult your provider's docs before running the bot, and don't expect public RPC nodes to hold up. +One case worth knowing about: `getProgramAccounts` over the whole pump.fun program is no longer served by anyone. That program owns more than 10 million accounts, so providers reject the request or time out no matter which filters you pass. Use a filtered subscription instead — `bonding-curve-progress/get_graduating_tokens.py` shows the pattern. + For Chainstack, the numbers you need are in the [throughput guidelines](https://docs.chainstack.com/docs/limits), kept up to date. The bot rate-limits itself with a token bucket: `node.max_rps` in the YAML (25 by default) smooths the request rate while allowing short bursts, and 429s are retried with exponential backoff. diff --git a/learning-examples/bonding-curve-progress/get_graduating_tokens.py b/learning-examples/bonding-curve-progress/get_graduating_tokens.py index 56fb4f3..06d2905 100644 --- a/learning-examples/bonding-curve-progress/get_graduating_tokens.py +++ b/learning-examples/bonding-curve-progress/get_graduating_tokens.py @@ -1,168 +1,476 @@ -""" -Module for querying and analyzing soon-to-gradute tokens in the Pump.fun program. -It includes functionality to fetch bonding curves based on token reserves and -find associated SPL token accounts. +"""Watch for pump.fun coins approaching graduation, over plain WebSocket RPC. -Note: getProgramAccounts may be slow as it is a pretty heavy method for RPC. +Usage: + uv run learning-examples/bonding-curve-progress/get_graduating_tokens.py + uv run learning-examples/bonding-curve-progress/get_graduating_tokens.py --min-progress 95 + +Why a subscription and not `getProgramAccounts`: the pump.fun program now owns +over 10 million accounts, and every provider refuses to scan it. Helius, Alchemy +and dRPC reject with `Too many accounts requested (10000001 pubkeys)`; QuickNode +and Chainstack time out. No filter set fixes that — the rejection is on program +size, before filters apply. `getProgramAccountsV2` is a provider extension (Helius, +Solana Tracker), not core Agave, and its `limit` is a *scan* budget rather than a +result count, so answering this question with it means ~1000 sequential pages. + +`programSubscribe` sidesteps the scan entirely. A curve can only approach +graduation by being traded, and every write to it pushes the full 151-byte account, +so each notification carries everything needed to compute progress — there is no +state to accumulate and no cold start beyond the next trade. Verified accepted on +both a paid endpoint and the public `api.mainnet-beta.solana.com`. + +See `get_graduating_tokens_geyser.py` for the same report over Geyser gRPC, which +also gives you the transaction signature behind each update. + +Selecting a graduation threshold +-------------------------------- +Progress is measured against `Global.initial_real_token_reserves` (~793.1M tokens) +read from chain, not a hardcoded constant, because a mayhem coin can launch with +different virtual params and would otherwise show the wrong percentage. + +The pre-filter the server applies can only match exact bytes, so it cannot do +"anything above 90%". It can only do a few fixed cutoffs. `--min-progress` uses the +closest cutoff that is still wide enough, then makes the exact comparison here: + + filter cutoff ≈ graduated past + 2 zero bytes @ 30 281.5M tokens left 64.5% + 3 zero bytes @ 29 1.1M tokens left 99.86% + 4 zero bytes @ 28 4,295 tokens left 99.9995% + + --min-progress pre-filtered by the server? + below 64.5% no, every curve arrives and is filtered here + 64.5% to 99.86% yes, at the 64.5% cutoff + 99.86% and up yes, at the 99.86% cutoff + +So the pre-filter saves traffic, it does not decide the answer — whatever percentage +you ask for is honoured either way. Low thresholds just cost more bandwidth. If you +want to hand-tune, pick a different cutoff from the table: the last moments before +migration want the 3-byte one, a wider funnel the 2-byte one. + +Checked against mainnet by running the filtered and unfiltered subscriptions side by +side for a minute: same curves, nothing dropped, nothing extra. + +`dataSize: 151` restricts this to the current curve layout. The original 49-byte +layout (no `creator` field) still has accounts with `complete = false`, but none of +them are written to any more — verified over a 45s window in which all 205 updates +across 24 curves were 151-byte accounts. """ +import argparse import asyncio +import base64 +import json import os import struct -from typing import Final +import sys +from typing import Any, Final +import websockets from dotenv import load_dotenv from solana.rpc.async_api import AsyncClient -from solana.rpc.types import MemcmpOpts, TokenAccountOpts +from solana.rpc.types import TokenAccountOpts from solders.pubkey import Pubkey load_dotenv() # Constants -RPC_ENDPOINT: Final[str] = os.environ.get("SOLANA_NODE_RPC_ENDPOINT") +RPC_ENDPOINT: Final[str] = os.environ.get("SOLANA_NODE_RPC_ENDPOINT", "") +WSS_ENDPOINT: Final[str] = os.environ.get("SOLANA_NODE_WSS_ENDPOINT", "") PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string( "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P" ) +PUMP_GLOBAL: Final[Pubkey] = Pubkey.from_string( + "4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf" +) + +# Coins created by `create_v2` are Token-2022, so that is tried first. Querying +# under the wrong token program returns nothing at all. +TOKEN_2022_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string( + "TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb" +) TOKEN_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string( "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" ) -# The 8-byte discriminator for bonding curve accounts in Pump.fun -BONDING_CURVE_DISCRIMINATOR_BYTES: Final[bytes] = bytes.fromhex("17b7f83760d8ac60") +# See learning-examples/calculate_discriminator.py +BONDING_CURVE_DISCRIMINATOR: Final[bytes] = bytes.fromhex("17b7f83760d8ac60") +CURVE_ACCOUNT_LEN: Final[int] = 151 + +TOKEN_DECIMALS: Final[int] = 6 +_RESERVES_OFFSET: Final[int] = 24 # real_token_reserves, u64 LE +_COMPLETE_OFFSET: Final[int] = 48 +_QUOTE_MINT_OFFSET: Final[int] = 83 +_GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET: Final[int] = 89 + +_BAD_DISCRIMINATOR_MSG: Final[str] = "Invalid discriminator for bonding curve" + +# Quote assets. `quote_mint` is all zeros on SOL-paired coins, and the quote-side +# reserves are in that mint's raw units — 1e9 for SOL, 1e6 for USDC. +DEFAULT_QUOTE_MINT: Final[Pubkey] = Pubkey.from_bytes(bytes(32)) +WSOL_MINT: Final[Pubkey] = Pubkey.from_string( + "So11111111111111111111111111111111111111112" +) +USDC_MINT: Final[Pubkey] = Pubkey.from_string( + "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" +) +QUOTE_DECIMALS: Final[dict[Pubkey, int]] = {WSOL_MINT: 9, USDC_MINT: 6} +QUOTE_SYMBOLS: Final[dict[Pubkey, str]] = {WSOL_MINT: "SOL", USDC_MINT: "USDC"} + +# 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 + +# A qualifying curve is traded several times a second. Reprint it only once it has +# moved this far, so the output stays readable. +REPRINT_STEP_PCT: Final[float] = 0.25 + +RECONNECT_DELAY: Final[int] = 5 -async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> list: - """ - Fetch bonding curve accounts with real token reserves below a threshold. +def zero_prefix_gate(bound_raw: int) -> tuple[int, bytes] | None: + """Pick the tightest server-side cutoff that still lets every match through. + + See the module docstring for the cutoffs on offer. Returns None when none of + them is wide enough to be safe, in which case there is no pre-filtering and + every curve is checked here instead. Args: - client: Optional AsyncClient instance. If None, a new one will be created. + bound_raw: Highest reserves value, in raw units, that should still qualify Returns: - List of bonding curve accounts matching the criteria + An (offset, zero bytes) pair for a memcmp filter, or None for no filter """ - # Define the reserve threshold (100 trillion in token base units) - threshold: int = 100_000_000_000_000 - threshold_bytes: bytes = threshold.to_bytes(8, "little") - msb_prefix: bytes = threshold_bytes[6:] # Most significant bytes for pre-filtering + # Only 2, 3 and 4 zero bytes are offered. One zero byte would be a cutoff so + # high that no coin could ever fail it, which filters nothing while looking + # like it does. + for zero_bytes in (4, 3, 2): + if 2 ** (8 * (8 - zero_bytes)) > bound_raw: + return 32 - zero_bytes, bytes(zero_bytes) + return None - should_close_client: bool = client is None + +def build_filters(bound_raw: int) -> list[dict[str, Any]]: + """Assemble the server-side `programSubscribe` filters. + + Args: + bound_raw: Highest qualifying `real_token_reserves`, in raw units + + Returns: + Filter dicts in the shape the RPC expects + """ + + def memcmp(offset: int, raw: bytes) -> dict[str, Any]: + return { + "memcmp": { + "offset": offset, + "bytes": base64.b64encode(raw).decode(), + "encoding": "base64", + } + } + + filters: list[dict[str, Any]] = [ + {"dataSize": CURVE_ACCOUNT_LEN}, + memcmp(0, BONDING_CURVE_DISCRIMINATOR), + memcmp(_COMPLETE_OFFSET, b"\x00"), # Not graduated yet + ] + + gate = zero_prefix_gate(bound_raw) + if gate: + filters.append(memcmp(*gate)) + return filters + + +def parse_curve(data: bytes) -> dict[str, Any]: + """Decode the 151-byte bonding curve fields needed for a progress report. + + Args: + data: Raw bonding curve account data + + Returns: + Reserves in whole tokens, plus the quote asset's symbol + + Raises: + ValueError: If the discriminator does not match a bonding curve + """ + if data[:8] != BONDING_CURVE_DISCRIMINATOR: + raise ValueError(_BAD_DISCRIMINATOR_MSG) + + real_token_reserves = struct.unpack_from(" float: + """Read the launch-time real token reserves from the Global account. + + Global layout up to this field: discriminator(8) + initialized(1) + + authority(32) + fee_recipient(32) + initial_virtual_token_reserves(8) + + initial_virtual_sol_reserves(8), so initial_real_token_reserves sits at 89. + + Args: + client: Connected RPC client + + Returns: + Initial real token reserves in whole tokens, or the fallback constant + """ try: - if should_close_client: - client = AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=180) - await client.is_connected() + resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64") + raw = struct.unpack_from( + " Pubkey | None: + """Recover a coin's mint from its bonding curve address. + + The curve account carries no mint field and `["bonding-curve", mint]` is not + reversible, so this goes through the associated bonding curve — an ordinary ATA + owned by the curve. That ATA belongs to Token-2022 for `create_v2` coins, which + is every coin now being launched, so Token-2022 is tried first. The answer is + checked by re-deriving the curve PDA from the mint. + + Args: + client: Connected RPC client + curve: The bonding curve address + + Returns: + The mint, or None if no owned token account resolves back to this curve + """ + for program_id in (TOKEN_2022_PROGRAM_ID, TOKEN_PROGRAM_ID): + try: + resp = await client.get_token_accounts_by_owner( + curve, TokenAccountOpts(program_id=program_id) + ) + except Exception as e: # noqa: BLE001 - a miss here is not fatal + print(f"⚠️ Mint lookup failed for {curve}: {e}") + continue + + if not resp.value: + continue + + mint = Pubkey(resp.value[0].account.data[:32]) + derived, _ = Pubkey.find_program_address( + [b"bonding-curve", bytes(mint)], PUMP_PROGRAM_ID + ) + if derived == curve: + return mint + + return None + + +def print_banner(baseline: float, min_progress: float) -> None: + """Describe the baseline and the filter that will be installed. + + Args: + baseline: Launch-time real token reserves, in whole tokens + min_progress: Graduation progress threshold, as a percentage + """ + print(f"Graduation baseline: {baseline:,.0f} tokens (from Global)") + print(f"Reporting curves at or above {min_progress:.2f}% graduated") + + gate = zero_prefix_gate(progress_to_bound(baseline, min_progress)) + if gate: + cutoff_tokens = 2 ** (8 * (8 - len(gate[1]))) / 10**TOKEN_DECIMALS + cutoff_pct = max(100 - cutoff_tokens * 100 / baseline, 0.0) + print( + f"Pre-filter: the server sends only curves past ~{cutoff_pct:.2f}% " + f"({cutoff_tokens:,.0f} tokens left); the rest is checked here" + ) + else: + print("Pre-filter: none, so every curve arrives and is checked here") + print("Waiting for trades on qualifying curves...\n") + + +def progress_to_bound(baseline: float, min_progress: float) -> int: + """Convert a progress threshold into a raw `real_token_reserves` ceiling. + + Args: + baseline: Launch-time real token reserves, in whole tokens + min_progress: Graduation progress threshold, as a percentage + + Returns: + The highest raw reserves value that still qualifies + """ + return int(baseline * (1 - min_progress / 100) * 10**TOKEN_DECIMALS) + + +class GraduationReporter: + """Turns raw curve updates into one printed line per meaningful change. + + Holds the mint cache and the last-printed progress per curve, so the transport + loop only has to hand over decoded account bytes. + """ + + def __init__( + self, client: AsyncClient, baseline: float, min_progress: float + ) -> None: + """Initialize the reporter. + + Args: + client: Connected RPC client, used to resolve mints + baseline: Launch-time real token reserves, in whole tokens + min_progress: Graduation progress threshold, as a percentage + """ + self.client = client + self.baseline = baseline + self.min_progress = min_progress + self.mints: dict[Pubkey, Pubkey | None] = {} + self.last_printed: dict[Pubkey, float] = {} + + async def handle(self, curve: Pubkey, data: bytes, suffix: str = "") -> None: + """Report one curve update, if it qualifies and has moved far enough. + + Args: + curve: The bonding curve address + data: Raw bonding curve account data + suffix: Extra provenance to append to the line, if the transport has any + """ + try: + state = parse_curve(data) + except (ValueError, struct.error) as e: + print(f"⚠️ Could not decode {curve}: {e}") + return + + # The server-side gate is coarser than the requested threshold, so the exact + # comparison happens here. + progress = max(100 - state["real_token_reserves"] * 100 / self.baseline, 0.0) + if progress < self.min_progress: + return + + previous = self.last_printed.get(curve) + if previous is not None and abs(progress - previous) < REPRINT_STEP_PCT: + return + self.last_printed[curve] = progress + + if curve not in self.mints: + self.mints[curve] = await resolve_mint(self.client, curve) + mint = self.mints[curve] + + print( + f"🎓 {progress:6.2f}% " + f"mint={mint if mint else ''} " + f"curve={curve} " + f"{state['real_token_reserves']:,.0f} tokens left " + f"{state['real_quote_reserves']:,.4f} {state['quote_symbol']}" + f"{suffix}" ) - result = [] - for acc in response.value: - raw = acc.account.data - # Extract real_token_reserves (u64 = 8 bytes, little-endian) - offset: int = 24 # real_token_reserves field offset - real_token_reserves: int = struct.unpack(" None: + """Subscribe and consume notifications until the connection drops. Args: - bonding_curve_address: The bonding curve public key (as a string) - client: Optional AsyncClient instance. If None, a new one will be created. + reporter: Sink for decoded curve updates + filters: Server-side filters for the subscription - Returns: - The associated SPL token account data or None if not found + Raises: + ConnectionRefusedError: If the endpoint rejects the subscription outright """ - should_close_client: bool = client is None - try: - if should_close_client: - client = AsyncClient(RPC_ENDPOINT) - await client.is_connected() - - response = await client.get_token_accounts_by_owner( - Pubkey.from_string(bonding_curve_address), - TokenAccountOpts(program_id=TOKEN_PROGRAM_ID), + async with websockets.connect(WSS_ENDPOINT, max_size=None) as ws: + await ws.send( + json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "method": "programSubscribe", + "params": [ + str(PUMP_PROGRAM_ID), + { + "encoding": "base64", + "commitment": "processed", + "filters": filters, + }, + ], + } + ) ) - if response.value and len(response.value) > 0: - return response.value[0].account - else: - print(f"No token accounts found for {bonding_curve_address}") - return None - except Exception as e: - print(f"Error finding associated token account: {e}") - return None - finally: - if should_close_client and client: - await client.close() + ack = json.loads(await ws.recv()) + if "error" in ack: + raise ConnectionRefusedError(str(ack["error"])) + while True: + # ConnectionClosed deliberately propagates to the reconnect handler in + # watch(). Swallowing it here would make every further recv() raise + # instantly, forever. A JSONDecodeError is per-message rather than + # per-connection, so that one is safe to skip. + try: + message = json.loads(await ws.recv()) + except json.JSONDecodeError: + continue -def get_mint_address(data: bytes) -> str: - """ - Extract the mint address from SPL token account data. + if message.get("method") != "programNotification": + continue - Args: - data: The token account data as bytes - - Returns: - The mint address as a base58-encoded string - """ - return str(Pubkey(data[:32])) - - -async def main() -> None: - """Main entry point for querying and processing bonding curves.""" - async with AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=240) as client: - await client.is_connected() - - bonding_curves = await get_bonding_curves_by_reserves(client) - print(f"Total matches: {len(bonding_curves)}") - print("=" * 50) - - for bonding_curve in bonding_curves: - # Find the SPL token account owned by the bonding curve - associated_token_account = await find_associated_bonding_curve( - str(bonding_curve.pubkey), client + value = message["params"]["result"]["value"] + await reporter.handle( + Pubkey.from_string(value["pubkey"]), + base64.b64decode(value["account"]["data"][0]), ) - if associated_token_account: - mint_address = get_mint_address(associated_token_account.data) - print(f"Bonding curve: {bonding_curve.pubkey}") - print(f"Mint address: {mint_address}") - print("=" * 50) - # For demonstration, only process the first curve - break +async def watch(min_progress: float) -> None: + """Stream curve updates and report coins at or above `min_progress`. + + Args: + min_progress: Graduation progress threshold, as a percentage + """ + if not WSS_ENDPOINT or not RPC_ENDPOINT: + print("❌ Set SOLANA_NODE_RPC_ENDPOINT and SOLANA_NODE_WSS_ENDPOINT in .env") + return + + async with AsyncClient(RPC_ENDPOINT) as client: + baseline = await fetch_initial_real_token_reserves(client) + print_banner(baseline, min_progress) + + filters = build_filters(progress_to_bound(baseline, min_progress)) + reporter = GraduationReporter(client, baseline, min_progress) + + while True: + try: + await stream_once(reporter, filters) + except ConnectionRefusedError as e: + print(f"❌ Subscription rejected: {e}") + return + except Exception as e: # noqa: BLE001 - keep watching across hiccups + print(f"⚠️ {type(e).__name__}: {e}; reconnecting in {RECONNECT_DELAY}s") + await asyncio.sleep(RECONNECT_DELAY) + + +def main() -> None: + """Parse arguments and start watching.""" + # A watcher is usually piped into a file or grep, where block buffering would + # hold every line back — and lose them entirely if the process is killed. + sys.stdout.reconfigure(line_buffering=True) + + parser = argparse.ArgumentParser( + description="Report pump.fun coins approaching graduation, over WebSocket RPC" + ) + parser.add_argument( + "--min-progress", + type=float, + default=90.0, + help="Only report curves at or above this graduation percentage " + "(default: 90.0)", + ) + args = parser.parse_args() + asyncio.run(watch(args.min_progress)) if __name__ == "__main__": - asyncio.run(main()) + main() diff --git a/learning-examples/bonding-curve-progress/get_graduating_tokens_geyser.py b/learning-examples/bonding-curve-progress/get_graduating_tokens_geyser.py new file mode 100644 index 0000000..d5944da --- /dev/null +++ b/learning-examples/bonding-curve-progress/get_graduating_tokens_geyser.py @@ -0,0 +1,497 @@ +"""Watch for pump.fun coins approaching graduation, over Geyser gRPC. + +Usage: + uv run learning-examples/bonding-curve-progress/get_graduating_tokens_geyser.py + uv run learning-examples/bonding-curve-progress/get_graduating_tokens_geyser.py --min-progress 95 + +Needs GEYSER_ENDPOINT, GEYSER_API_TOKEN and GEYSER_AUTH_TYPE in .env, plus +SOLANA_NODE_RPC_ENDPOINT for the two things the stream cannot answer: the Global +baseline and each coin's mint. Geyser is a paid add-on, so `get_graduating_tokens.py` +is the portable version of this report — it runs on any endpoint, including the public +one. This variant exists because Geyser is common among traders and gives you the slot +and transaction signature behind every update, which the WebSocket feed does not. + +Why a subscription and not `getProgramAccounts`: the pump.fun program now owns over +10 million accounts, and every provider refuses to scan it — the rejection is on +program size, before filters apply. A curve can only approach graduation by being +traded, and every write pushes the full 151-byte account, so each update carries +everything needed to compute progress: no accumulated state, no cold start beyond the +next trade. + +Selecting a graduation threshold +-------------------------------- +Progress is measured against `Global.initial_real_token_reserves` (~793.1M tokens) +read from chain, not a hardcoded constant, because a mayhem coin can launch with +different virtual params and would otherwise show the wrong percentage. + +The pre-filter the server applies can only match exact bytes, so it cannot do +"anything above 90%". It can only do a few fixed cutoffs. `--min-progress` uses the +closest cutoff that is still wide enough, then makes the exact comparison here: + + filter cutoff ≈ graduated past + 2 zero bytes @ 30 281.5M tokens left 64.5% + 3 zero bytes @ 29 1.1M tokens left 99.86% + 4 zero bytes @ 28 4,295 tokens left 99.9995% + + --min-progress pre-filtered by the server? + below 64.5% no, every curve arrives and is filtered here + 64.5% to 99.86% yes, at the 64.5% cutoff + 99.86% and up yes, at the 99.86% cutoff + +So the pre-filter saves traffic, it does not decide the answer — whatever percentage +you ask for is honoured either way. Low thresholds just cost more bandwidth. If you +want to hand-tune, pick a different cutoff from the table: the last moments before +migration want the 3-byte one, a wider funnel the 2-byte one. + +Checked against mainnet by running the filtered and unfiltered subscriptions side by +side for a minute: same curves, nothing dropped, nothing extra. + +`datasize = 151` restricts this to the current curve layout. The original 49-byte +layout still has accounts with `complete = false`, but none of them are written to any +more — verified over a 45s window in which all 205 updates across 24 curves were +151-byte accounts. +""" + +import argparse +import asyncio +import os +import struct +import sys +from pathlib import Path +from typing import Any, Final + +import grpc +from dotenv import load_dotenv +from solana.rpc.async_api import AsyncClient +from solana.rpc.types import TokenAccountOpts +from solders.pubkey import Pubkey +from solders.signature import Signature + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) +from src.geyser.generated import ( + geyser_pb2, + geyser_pb2_grpc, +) + +load_dotenv() + +# Constants +RPC_ENDPOINT: Final[str] = os.environ.get("SOLANA_NODE_RPC_ENDPOINT", "") +GEYSER_ENDPOINT: Final[str] = os.environ.get("GEYSER_ENDPOINT", "") +GEYSER_API_TOKEN: Final[str] = os.environ.get("GEYSER_API_TOKEN", "") +GEYSER_AUTH_TYPE: Final[str] = os.environ.get("GEYSER_AUTH_TYPE", "x-token").lower() + +PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string( + "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P" +) +PUMP_GLOBAL: Final[Pubkey] = Pubkey.from_string( + "4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf" +) + +# Coins created by `create_v2` are Token-2022, so that is tried first. Querying under +# the wrong token program returns nothing at all. +TOKEN_2022_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string( + "TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb" +) +TOKEN_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string( + "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" +) + +# See learning-examples/calculate_discriminator.py +BONDING_CURVE_DISCRIMINATOR: Final[bytes] = bytes.fromhex("17b7f83760d8ac60") +CURVE_ACCOUNT_LEN: Final[int] = 151 + +TOKEN_DECIMALS: Final[int] = 6 +_RESERVES_OFFSET: Final[int] = 24 # real_token_reserves, u64 LE +_COMPLETE_OFFSET: Final[int] = 48 +_QUOTE_MINT_OFFSET: Final[int] = 83 +_GLOBAL_INITIAL_REAL_TOKEN_RESERVES_OFFSET: Final[int] = 89 + +_BAD_DISCRIMINATOR_MSG: Final[str] = "Invalid discriminator for bonding curve" +_BAD_AUTH_TYPE_MSG: Final[str] = "GEYSER_AUTH_TYPE must be 'x-token' or 'basic'" + +# Quote assets. `quote_mint` is all zeros on SOL-paired coins, and the quote-side +# reserves are in that mint's raw units — 1e9 for SOL, 1e6 for USDC. +DEFAULT_QUOTE_MINT: Final[Pubkey] = Pubkey.from_bytes(bytes(32)) +WSOL_MINT: Final[Pubkey] = Pubkey.from_string( + "So11111111111111111111111111111111111111112" +) +USDC_MINT: Final[Pubkey] = Pubkey.from_string( + "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" +) +QUOTE_DECIMALS: Final[dict[Pubkey, int]] = {WSOL_MINT: 9, USDC_MINT: 6} +QUOTE_SYMBOLS: Final[dict[Pubkey, str]] = {WSOL_MINT: "SOL", USDC_MINT: "USDC"} + +# 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 + +# A qualifying curve is traded several times a second. Reprint it only once it has +# moved this far, so the output stays readable. +REPRINT_STEP_PCT: Final[float] = 0.25 + +RECONNECT_DELAY: Final[int] = 5 + + +def zero_prefix_gate(bound_raw: int) -> tuple[int, bytes] | None: + """Pick the tightest server-side cutoff that still lets every match through. + + See the module docstring for the cutoffs on offer. Returns None when none of + them is wide enough to be safe, in which case there is no pre-filtering and + every curve is checked here instead. + + Args: + bound_raw: Highest reserves value, in raw units, that should still qualify + + Returns: + An (offset, zero bytes) pair for a memcmp filter, or None for no filter + """ + # Only 2, 3 and 4 zero bytes are offered. One zero byte would be a cutoff so + # high that no coin could ever fail it, which filters nothing while looking + # like it does. + for zero_bytes in (4, 3, 2): + if 2 ** (8 * (8 - zero_bytes)) > bound_raw: + return 32 - zero_bytes, bytes(zero_bytes) + return None + + +def build_subscribe_request(bound_raw: int) -> geyser_pb2.SubscribeRequest: + """Build the Geyser account subscription for near-graduation curves. + + Args: + bound_raw: Highest qualifying `real_token_reserves`, in raw units + + Returns: + The subscription request + """ + request = geyser_pb2.SubscribeRequest() + accounts = request.accounts["graduating_curves"] + accounts.owner.append(str(PUMP_PROGRAM_ID)) + + accounts.filters.add().datasize = CURVE_ACCOUNT_LEN + + discriminator = accounts.filters.add().memcmp + discriminator.offset = 0 + discriminator.bytes = BONDING_CURVE_DISCRIMINATOR + + not_complete = accounts.filters.add().memcmp + not_complete.offset = _COMPLETE_OFFSET + not_complete.bytes = b"\x00" # Not graduated yet + + gate = zero_prefix_gate(bound_raw) + if gate: + reserves = accounts.filters.add().memcmp + reserves.offset, reserves.bytes = gate + + request.commitment = geyser_pb2.CommitmentLevel.PROCESSED + return request + + +def create_geyser_connection() -> tuple[Any, grpc.aio.Channel]: + """Open an authenticated gRPC channel to the Geyser endpoint. + + Returns: + The Geyser stub and the channel backing it + + Raises: + ValueError: If GEYSER_AUTH_TYPE is not a supported scheme + """ + if GEYSER_AUTH_TYPE == "x-token": + auth = grpc.metadata_call_credentials( + lambda _, callback: callback((("x-token", GEYSER_API_TOKEN),), None) + ) + elif GEYSER_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) + endpoint = ( + GEYSER_ENDPOINT.replace("https://", "").replace("http://", "").rstrip("/") + ) + channel = grpc.aio.secure_channel(endpoint, creds) + return geyser_pb2_grpc.GeyserStub(channel), channel + + +def parse_curve(data: bytes) -> dict[str, Any]: + """Decode the 151-byte bonding curve fields needed for a progress report. + + Args: + data: Raw bonding curve account data + + Returns: + Reserves in whole tokens, plus the quote asset's symbol + + Raises: + ValueError: If the discriminator does not match a bonding curve + """ + if data[:8] != BONDING_CURVE_DISCRIMINATOR: + raise ValueError(_BAD_DISCRIMINATOR_MSG) + + real_token_reserves = struct.unpack_from(" float: + """Read the launch-time real token reserves from the Global account. + + Global layout up to this field: discriminator(8) + initialized(1) + + authority(32) + fee_recipient(32) + initial_virtual_token_reserves(8) + + initial_virtual_sol_reserves(8), so initial_real_token_reserves sits at 89. + + Args: + client: Connected RPC client + + Returns: + Initial real token reserves in whole tokens, or the fallback constant + """ + try: + resp = await client.get_account_info(PUMP_GLOBAL, encoding="base64") + raw = struct.unpack_from( + " Pubkey | None: + """Recover a coin's mint from its bonding curve address. + + The curve account carries no mint field and `["bonding-curve", mint]` is not + reversible, so this goes through the associated bonding curve — an ordinary ATA + owned by the curve. That ATA belongs to Token-2022 for `create_v2` coins, which is + every coin now being launched, so Token-2022 is tried first. The answer is checked + by re-deriving the curve PDA from the mint. + + Args: + client: Connected RPC client + curve: The bonding curve address + + Returns: + The mint, or None if no owned token account resolves back to this curve + """ + for program_id in (TOKEN_2022_PROGRAM_ID, TOKEN_PROGRAM_ID): + try: + resp = await client.get_token_accounts_by_owner( + curve, TokenAccountOpts(program_id=program_id) + ) + except Exception as e: # noqa: BLE001 - a miss here is not fatal + print(f"⚠️ Mint lookup failed for {curve}: {e}") + continue + + if not resp.value: + continue + + mint = Pubkey(resp.value[0].account.data[:32]) + derived, _ = Pubkey.find_program_address( + [b"bonding-curve", bytes(mint)], PUMP_PROGRAM_ID + ) + if derived == curve: + return mint + + return None + + +def progress_to_bound(baseline: float, min_progress: float) -> int: + """Convert a progress threshold into a raw `real_token_reserves` ceiling. + + Args: + baseline: Launch-time real token reserves, in whole tokens + min_progress: Graduation progress threshold, as a percentage + + Returns: + The highest raw reserves value that still qualifies + """ + return int(baseline * (1 - min_progress / 100) * 10**TOKEN_DECIMALS) + + +def print_banner(baseline: float, min_progress: float) -> None: + """Describe the baseline and the filter that will be installed. + + Args: + baseline: Launch-time real token reserves, in whole tokens + min_progress: Graduation progress threshold, as a percentage + """ + print(f"Graduation baseline: {baseline:,.0f} tokens (from Global)") + print(f"Reporting curves at or above {min_progress:.2f}% graduated") + + gate = zero_prefix_gate(progress_to_bound(baseline, min_progress)) + if gate: + cutoff_tokens = 2 ** (8 * (8 - len(gate[1]))) / 10**TOKEN_DECIMALS + cutoff_pct = max(100 - cutoff_tokens * 100 / baseline, 0.0) + print( + f"Pre-filter: the server sends only curves past ~{cutoff_pct:.2f}% " + f"({cutoff_tokens:,.0f} tokens left); the rest is checked here" + ) + else: + print("Pre-filter: none, so every curve arrives and is checked here") + print("Waiting for trades on qualifying curves...\n") + + +class GraduationReporter: + """Turns raw curve updates into one printed line per meaningful change. + + Holds the mint cache and the last-printed progress per curve, so the transport + loop only has to hand over decoded account bytes. + """ + + def __init__( + self, client: AsyncClient, baseline: float, min_progress: float + ) -> None: + """Initialize the reporter. + + Args: + client: Connected RPC client, used to resolve mints + baseline: Launch-time real token reserves, in whole tokens + min_progress: Graduation progress threshold, as a percentage + """ + self.client = client + self.baseline = baseline + self.min_progress = min_progress + self.mints: dict[Pubkey, Pubkey | None] = {} + self.last_printed: dict[Pubkey, float] = {} + + async def handle(self, curve: Pubkey, data: bytes, suffix: str = "") -> None: + """Report one curve update, if it qualifies and has moved far enough. + + Args: + curve: The bonding curve address + data: Raw bonding curve account data + suffix: Extra provenance to append to the line + """ + try: + state = parse_curve(data) + except (ValueError, struct.error) as e: + print(f"⚠️ Could not decode {curve}: {e}") + return + + # The server-side gate is coarser than the requested threshold, so the exact + # comparison happens here. + progress = max(100 - state["real_token_reserves"] * 100 / self.baseline, 0.0) + if progress < self.min_progress: + return + + previous = self.last_printed.get(curve) + if previous is not None and abs(progress - previous) < REPRINT_STEP_PCT: + return + self.last_printed[curve] = progress + + if curve not in self.mints: + self.mints[curve] = await resolve_mint(self.client, curve) + mint = self.mints[curve] + + print( + f"🎓 {progress:6.2f}% " + f"mint={mint if mint else ''} " + f"curve={curve} " + f"{state['real_token_reserves']:,.0f} tokens left " + f"{state['real_quote_reserves']:,.4f} {state['quote_symbol']}" + f"{suffix}" + ) + + +async def stream_once(reporter: GraduationReporter, bound_raw: int) -> None: + """Subscribe over Geyser and consume account updates until the stream ends. + + Args: + reporter: Sink for decoded curve updates + bound_raw: Highest qualifying `real_token_reserves`, in raw units + """ + stub, channel = create_geyser_connection() + try: + request = build_subscribe_request(bound_raw) + async for update in stub.Subscribe(iter([request])): + if not update.HasField("account"): + continue + + account = update.account.account + signature = ( + str(Signature(bytes(account.txn_signature))) + if account.txn_signature + else "" + ) + await reporter.handle( + Pubkey.from_bytes(bytes(account.pubkey)), + bytes(account.data), + suffix=f" slot={update.account.slot} sig={signature}", + ) + finally: + await channel.close() + + +async def watch(min_progress: float) -> None: + """Stream curve updates over Geyser and report coins at or above `min_progress`. + + Args: + min_progress: Graduation progress threshold, as a percentage + """ + if not GEYSER_ENDPOINT or not GEYSER_API_TOKEN: + print("❌ Set GEYSER_ENDPOINT and GEYSER_API_TOKEN in .env") + return + if not RPC_ENDPOINT: + print("❌ Set SOLANA_NODE_RPC_ENDPOINT in .env (needed for Global and mints)") + return + + async with AsyncClient(RPC_ENDPOINT) as client: + baseline = await fetch_initial_real_token_reserves(client) + print_banner(baseline, min_progress) + + bound_raw = progress_to_bound(baseline, min_progress) + reporter = GraduationReporter(client, baseline, min_progress) + + while True: + try: + await stream_once(reporter, bound_raw) + except ValueError as e: + print(f"❌ {e}") + return + except grpc.aio.AioRpcError as e: + print( + f"⚠️ gRPC error ({e.code()}): {e.details()}; " + f"reconnecting in {RECONNECT_DELAY}s" + ) + await asyncio.sleep(RECONNECT_DELAY) + except Exception as e: # noqa: BLE001 - keep watching across hiccups + print(f"⚠️ {type(e).__name__}: {e}; reconnecting in {RECONNECT_DELAY}s") + await asyncio.sleep(RECONNECT_DELAY) + + +def main() -> None: + """Parse arguments and start watching.""" + # A watcher is usually piped into a file or grep, where block buffering would + # hold every line back — and lose them entirely if the process is killed. + sys.stdout.reconfigure(line_buffering=True) + + parser = argparse.ArgumentParser( + description="Report pump.fun coins approaching graduation, over Geyser gRPC" + ) + parser.add_argument( + "--min-progress", + type=float, + default=90.0, + help="Only report curves at or above this graduation percentage " + "(default: 90.0)", + ) + args = parser.parse_args() + asyncio.run(watch(args.min_progress)) + + +if __name__ == "__main__": + main()