Add bonding curve state check & Raydium migration event listening

learning-examples/check_boding_curve_status.py

@@ -0,0 +1,96 @@
+import asyncio
+import struct
+from typing import Final
+from construct import Struct, Int64ul, Flag
+from solana.rpc.async_api import AsyncClient
+from solders.pubkey import Pubkey
+import argparse
+import sys
+import os
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+from config import RPC_ENDPOINT, PUMP_PROGRAM
+
+# Constants
+EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)
+
+class BondingCurveState:
+    _STRUCT = Struct(
+        "virtual_token_reserves" / Int64ul,
+        "virtual_sol_reserves" / Int64ul,
+        "real_token_reserves" / Int64ul,
+        "real_sol_reserves" / Int64ul,
+        "token_total_supply" / Int64ul,
+        "complete" / Flag
+    )
+
+    def __init__(self, data: bytes) -> None:
+        parsed = self._STRUCT.parse(data[8:])
+        self.__dict__.update(parsed)
+
+def get_associated_bonding_curve_address(mint: Pubkey, program_id: Pubkey) -> tuple[Pubkey, int]:
+    """
+    Derives the associated bonding curve address for a given mint
+    """
+    return Pubkey.find_program_address(
+        [
+            b"bonding-curve",
+            bytes(mint)
+        ],
+        program_id
+    )
+
+async def get_bonding_curve_state(conn: AsyncClient, curve_address: Pubkey) -> BondingCurveState:
+    response = await conn.get_account_info(curve_address)
+    if not response.value or not response.value.data:
+        raise ValueError("Invalid curve state: No data")
+
+    data = response.value.data
+    if data[:8] != EXPECTED_DISCRIMINATOR:
+        raise ValueError("Invalid curve state discriminator")
+
+    return BondingCurveState(data)
+
+async def check_token_status(mint_address: str) -> None:
+    try:
+        mint = Pubkey.from_string(mint_address)
+        
+        # Get the associated bonding curve address
+        bonding_curve_address, bump = get_associated_bonding_curve_address(mint, PUMP_PROGRAM)
+        
+        print("\nToken Status:")
+        print("-" * 50)
+        print(f"Token Mint:              {mint}")
+        print(f"Associated Bonding Curve: {bonding_curve_address}")
+        print(f"Bump Seed:               {bump}")
+        print("-" * 50)
+        
+        # Check completion status
+        async with AsyncClient(RPC_ENDPOINT) as client:
+            try:
+                curve_state = await get_bonding_curve_state(client, bonding_curve_address)
+                
+                print("\nBonding Curve Status:")
+                print("-" * 50)
+                print(f"Completion Status: {'Completed' if curve_state.complete else 'Not Completed'}")
+                if curve_state.complete:
+                    print("\nNote: This bonding curve has completed and liquidity has been migrated to Raydium.")
+                print("-" * 50)
+                
+            except ValueError as e:
+                print(f"\nError accessing bonding curve: {e}")
+            
+    except ValueError as e:
+        print(f"\nError: Invalid address format - {e}")
+    except Exception as e:
+        print(f"\nUnexpected error: {e}")
+
+def main():
+    parser = argparse.ArgumentParser(description='Check token bonding curve status')
+    parser.add_argument('mint_address', help='The token mint address')
+    
+    args = parser.parse_args()
+    asyncio.run(check_token_status(args.mint_address))
+
+if __name__ == "__main__":
+    main()