Merge pull request #89 from chainstacklabs/feat/ps-migration-listener-and-other

Add programSubscribe listener for migrations and other (learning example)
2026-07-27 15:27:44 +00:00 · 2025-04-22 06:59:52 +00:00
parent 188f43f4cd 81c2af4af1
commit 2aac2f8f64
24 changed files with 1148 additions and 255 deletions
@@ -1,25 +1,42 @@
-import argparse
+"""
+Module for checking the status of a token's bonding curve on the Solana network using
+the Pump.fun program. It allows querying the bonding curve state and completion status.
+"""
+
 import asyncio
 import os
 import struct
-import sys
 from typing import Final

 from construct import Flag, Int64ul, Struct
+from dotenv import load_dotenv
 from solana.rpc.async_api import AsyncClient
 from solders.pubkey import Pubkey

-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-
-from core.pubkeys import PumpAddresses
-
-# Constants
-EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)
+load_dotenv()

 RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")

+# Change to token you want to query
+TOKEN_MINT = "xWrzYY4c1LnbSkLrd2LDUg9vw7YtVyJhGmw7MABpump"
+
+# Constants
+PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
+EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)
+

 class BondingCurveState:
+    """
+    Represents the state of a bonding curve account.
+    
+    Attributes:
+        virtual_token_reserves: Virtual token reserves in the curve
+        virtual_sol_reserves: Virtual SOL reserves in the curve
+        real_token_reserves: Real token reserves in the curve
+        real_sol_reserves: Real SOL reserves in the curve
+        token_total_supply: Total token supply in the curve
+        complete: Whether the curve has completed and liquidity migrated
+    """
    _STRUCT = Struct(
        "virtual_token_reserves" / Int64ul,
        "virtual_sol_reserves" / Int64ul,
@@ -38,7 +55,14 @@ def get_associated_bonding_curve_address(
    mint: Pubkey, program_id: Pubkey
 ) -> tuple[Pubkey, int]:
    """
-    Derives the associated bonding curve address for a given mint
+    Derives the associated bonding curve address for a given mint.
+    
+    Args:
+        mint: The token mint address
+        program_id: The program ID for the bonding curve
+        
+    Returns:
+        Tuple of (bonding curve address, bump seed)
    """
    return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)

@@ -46,6 +70,19 @@ def get_associated_bonding_curve_address(
 async def get_bonding_curve_state(
    conn: AsyncClient, curve_address: Pubkey
 ) -> BondingCurveState:
+    """
+    Fetches and validates the state of a bonding curve account.
+    
+    Args:
+        conn: AsyncClient connection to Solana RPC
+        curve_address: Address of the bonding curve account
+        
+    Returns:
+        BondingCurveState object containing parsed account data
+        
+    Raises:
+        ValueError: If account data is invalid or missing
+    """
    response = await conn.get_account_info(curve_address, encoding="base64")
    if not response.value or not response.value.data:
        raise ValueError("Invalid curve state: No data")
@@ -58,19 +95,25 @@ async def get_bonding_curve_state(


 async def check_token_status(mint_address: str) -> None:
+    """
+    Checks and prints the status of a token and its bonding curve.
+    
+    Args:
+        mint_address: The token mint address as a string
+    """
    try:
        mint = Pubkey.from_string(mint_address)

        # Get the associated bonding curve address
        bonding_curve_address, bump = get_associated_bonding_curve_address(
-            mint, PumpAddresses.PROGRAM
+            mint, PUMP_PROGRAM_ID
        )

-        print("\nToken Status:")
+        print("\nToken status:")
        print("-" * 50)
-        print(f"Token Mint:              {mint}")
-        print(f"Associated Bonding Curve: {bonding_curve_address}")
-        print(f"Bump Seed:               {bump}")
+        print(f"Token mint:              {mint}")
+        print(f"Associated bonding curve: {bonding_curve_address}")
+        print(f"Bump seed:               {bump}")
        print("-" * 50)

        # Check completion status
@@ -80,14 +123,14 @@ async def check_token_status(mint_address: str) -> None:
                    client, bonding_curve_address
                )

-                print("\nBonding Curve Status:")
+                print("\nBonding curve status:")
                print("-" * 50)
                print(
-                    f"Completion Status: {'Completed' if curve_state.complete else 'Not Completed'}"
+                    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."
+                        "\nNote: This bonding curve has completed and liquidity has been migrated to PumpSwap."
                    )
                print("-" * 50)

@@ -100,12 +143,13 @@ async def check_token_status(mint_address: str) -> None:
        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))
+def main() -> None:
+    """Main entry point for the token status checker."""
+    #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(TOKEN_MINT))


 if __name__ == "__main__":
@@ -1,6 +1,15 @@
+"""
+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.
+
+Note: getProgramAccounts may be slow as it is a pretty heavy method for RPC.
+"""
+
 import asyncio
 import os
 import struct
+from typing import Final

 from dotenv import load_dotenv
 from solana.rpc.async_api import AsyncClient
@@ -9,19 +18,18 @@ from solders.pubkey import Pubkey

 load_dotenv()

-RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT1")
-PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
-TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
+# Constants
+RPC_ENDPOINT: Final[str] = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
+PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
+TOKEN_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")

 # The 8-byte discriminator for bonding curve accounts in Pump.fun
-# Used to identify account types in getProgramAccounts requests
-BONDING_CURVE_DISCRIMINATOR_BYTES = bytes.fromhex("17b7f83760d8ac60")
+BONDING_CURVE_DISCRIMINATOR_BYTES: Final[bytes] = bytes.fromhex("17b7f83760d8ac60")


 async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> list:
    """
-    Fetch bonding curve accounts from the Pump.fun program that have real_token_reserves
-    below a defined threshold and are not marked as complete (i.e., not migrated).
+    Fetch bonding curve accounts with real token reserves below a threshold.
    
    Args:
        client: Optional AsyncClient instance. If None, a new one will be created.
@@ -29,31 +37,22 @@ async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> l
    Returns:
        List of bonding curve accounts matching the criteria
    """
-    # Define the reserve threshold we're interested in
-    # 100 trillion (in token base units)
-    threshold = 100_000_000_000_000
-    
-    # Convert the threshold into 8-byte little-endian format (as stored on-chain)
-    threshold_bytes = threshold.to_bytes(8, 'little')
-    
-    # Extract the 2 most significant bytes to pre-filter values less than 2^48 (~281T)
-    # This optimization reduces the number of accounts returned by the RPC
-    msb_prefix = threshold_bytes[6:]
+    # 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

-    should_close_client = client is None
+    should_close_client: bool = client is None
    try:
        if should_close_client:
-            client = AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=120)
+            client = AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=180)
            await client.is_connected()
            
        # Define on-chain filters for getProgramAccounts
        filters = [
-            # Match only bonding curve accounts
-            MemcmpOpts(offset=0, bytes=BONDING_CURVE_DISCRIMINATOR_BYTES),
-            # Real token reserves MSB bytes (pre-filter)
-            MemcmpOpts(offset=30, bytes=msb_prefix),
-            # Complete flag is False (not migrated)
-            MemcmpOpts(offset=48, bytes=b'\x00'),
+            MemcmpOpts(offset=0, bytes=BONDING_CURVE_DISCRIMINATOR_BYTES),  # Match bonding curve accounts
+            MemcmpOpts(offset=30, bytes=msb_prefix),  # Pre-filter by real token reserves MSB
+            MemcmpOpts(offset=48, bytes=b"\x00"),  # Ensure complete flag is False
        ]

        # Query accounts matching filters
@@ -67,26 +66,15 @@ async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> l
        for acc in response.value:
            raw = acc.account.data

-            # Parse account data according to the Pump.fun bonding curve layout:
-            # [8] discriminator
-            # [8] virtual_token_reserves (u64)
-            # [8] virtual_sol_reserves (u64)
-            # [8] real_token_reserves (u64)
-            # [8] real_sol_reserves (u64)
-            # [8] token_total_supply (u64)
-            # [1] complete (bool)
-            
-            # Skip to real_token_reserves field (8 + 8 + 8 = 24 bytes offset)
-            offset = 24
-            
            # Extract real_token_reserves (u64 = 8 bytes, little-endian)
-            real_token_reserves = struct.unpack("<Q", raw[offset:offset + 8])[0]
+            offset: int = 24  # real_token_reserves field offset
+            real_token_reserves: int = struct.unpack("<Q", raw[offset:offset + 8])[0]

-            # Post-filter: ensure value is really below the defined threshold
+            # Post-filter: ensure value is below the threshold
            if real_token_reserves < threshold:
                print(f"Pubkey: {acc.pubkey}")
                print(f"Real token reserves: {real_token_reserves / 10**6} tokens")
-                print("="*50)
+                print("=" * 50)
                result.append(acc)

        return result
@@ -95,13 +83,12 @@ async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> l
            await client.close()


-async def find_associated_bonding_curve(bonding_curve_address: str, client: AsyncClient | None = None):
+async def find_associated_bonding_curve(
+    bonding_curve_address: str, client: AsyncClient | None = None
+) -> dict | None:
    """
    Find the SPL token account owned by a bonding curve.
    
-    A bonding curve typically owns exactly one SPL token account that represents
-    the token being traded through the curve.
-    
    Args:
        bonding_curve_address: The bonding curve public key (as a string)
        client: Optional AsyncClient instance. If None, a new one will be created.
@@ -109,7 +96,7 @@ async def find_associated_bonding_curve(bonding_curve_address: str, client: Asyn
    Returns:
        The associated SPL token account data or None if not found
    """
-    should_close_client = client is None
+    should_close_client: bool = client is None
    try:
        if should_close_client:
            client = AsyncClient(RPC_ENDPOINT)
@@ -137,26 +124,23 @@ def get_mint_address(data: bytes) -> str:
    """
    Extract the mint address from SPL token account data.
    
-    In SPL token account data, the mint address is stored in the first 32 bytes.
-    
    Args:
        data: The token account data as bytes
        
    Returns:
        The mint address as a base58-encoded string
    """
-    # The mint address is stored in the first 32 bytes
-    mint_bytes = data[:32]
-    return str(Pubkey(mint_bytes))
+    return str(Pubkey(data[:32]))


-async def main():
+async def main() -> None:
+    """Main entry point for querying and processing bonding curves."""
    async with AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=120) as client:
        await client.is_connected()
        
        bonding_curves = await get_bonding_curves_by_reserves(client)
        print(f"Total matches: {len(bonding_curves)}")
-        print("="*50)
+        print("=" * 50)

        for bonding_curve in bonding_curves:
            # Find the SPL token account owned by the bonding curve
@@ -168,7 +152,7 @@ async def main():
                mint_address = get_mint_address(associated_token_account.data)
                print(f"Bonding curve: {bonding_curve.pubkey}")
                print(f"Mint address: {mint_address}")
-                print("="*50)
+                print("=" * 50)
            
            # For demonstration, only process the first curve
            break
@@ -0,0 +1,135 @@
+"""
+Module for tracking the progress of a bonding curve for a Pump.fun token.
+It continuously polls the bonding curve state and prints updates at regular intervals.
+"""
+
+import asyncio
+import os
+import struct
+from typing import Final
+
+from dotenv import load_dotenv
+from solana.rpc.async_api import AsyncClient
+from solders.pubkey import Pubkey
+
+load_dotenv()
+
+# Constants
+RPC_URL: Final[str] = os.getenv("SOLANA_NODE_RPC_ENDPOINT")
+TOKEN_MINT: Final[str] = "xWrzYY4c1LnbSkLrd2LDUg9vw7YtVyJhGmw7MABpump"
+PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
+LAMPORTS_PER_SOL: Final[int] = 1_000_000_000
+TOKEN_DECIMALS: Final[int] = 6
+EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 6966180631402821399)  # Pump.fun bonding curve discriminator
+POLL_INTERVAL: Final[int] = 10  # Seconds between each status check
+
+
+def get_associated_bonding_curve_address(mint: Pubkey, program_id: Pubkey) -> Pubkey:
+    """
+    Derive the bonding curve PDA address from a mint address.
+    
+    Args:
+        mint: The token mint address
+        program_id: The program ID for the bonding curve
+        
+    Returns:
+        The bonding curve address
+    """
+    return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)[0]
+
+
+async def get_account_data(client: AsyncClient, pubkey: Pubkey) -> bytes:
+    """
+    Fetch raw account data for a given public key.
+    
+    Args:
+        client: AsyncClient connection to Solana RPC
+        pubkey: The public key of the account to fetch
+        
+    Returns:
+        The raw account data as bytes
+        
+    Raises:
+        ValueError: If the account is not found or has no data
+    """
+    resp = await client.get_account_info(pubkey, encoding="base64")
+    if not resp.value or not resp.value.data:
+        raise ValueError(f"Account {pubkey} not found or has no data")
+
+    return resp.value.data
+
+
+def parse_curve_state(data: bytes) -> dict:
+    """
+    Decode bonding curve account data into a readable format.
+    
+    Args:
+        data: The raw bonding curve account data
+        
+    Returns:
+        A dictionary containing parsed bonding curve fields
+        
+    Raises:
+        ValueError: If the account discriminator is invalid
+    """
+    if data[:8] != EXPECTED_DISCRIMINATOR:
+        raise ValueError("Invalid discriminator for bonding curve")
+
+    fields = struct.unpack_from("<QQQQQ?", data, 8)
+    return {
+        "virtual_token_reserves": fields[0] / 10**TOKEN_DECIMALS,
+        "virtual_sol_reserves": fields[1] / LAMPORTS_PER_SOL,
+        "real_token_reserves": fields[2] / 10**TOKEN_DECIMALS,
+        "real_sol_reserves": fields[3] / LAMPORTS_PER_SOL,
+        "token_total_supply": fields[4] / 10**TOKEN_DECIMALS,
+        "complete": fields[5],
+    }
+
+
+def print_curve_status(state: dict) -> None:
+    """
+    Print the current status of the bonding curve in a readable format.
+    
+    Args:
+        state: The parsed bonding curve state dictionary
+    """
+    progress = 0
+    if state["token_total_supply"]:
+        progress = 100 - (100 * state["real_token_reserves"] / state["token_total_supply"])
+
+    print("=" * 30)
+    print(f"Complete: {'✅' if state['complete'] else '❌'}")
+    print(f"Progress: {progress:.2f}%")
+    print(f"Token reserves: {state['real_token_reserves']:.4f}")
+    print(f"SOL reserves:   {state['real_sol_reserves']:.4f}")
+    print("=" * 30, "\n")
+
+
+async def track_curve() -> None:
+    """
+    Continuously track and display the state of a bonding curve.
+    """
+    if not RPC_URL or not TOKEN_MINT:
+        print("❌ Set SOLANA_NODE_RPC_ENDPOINT and TOKEN_MINT in .env")
+        return
+
+    mint_pubkey: Pubkey = Pubkey.from_string(TOKEN_MINT)
+    curve_pubkey: Pubkey = get_associated_bonding_curve_address(mint_pubkey, PUMP_PROGRAM_ID)
+
+    print("Tracking bonding curve for:", mint_pubkey)
+    print("Curve address:", curve_pubkey, "\n")
+
+    async with AsyncClient(RPC_URL) as client:
+        while True:
+            try:
+                data = await get_account_data(client, curve_pubkey)
+                state = parse_curve_state(data)
+                print_curve_status(state)
+            except Exception as e:
+                print(f"⚠️ Error: {e}")
+
+            await asyncio.sleep(POLL_INTERVAL)
+
+
+if __name__ == "__main__":
+    asyncio.run(track_curve())
@@ -0,0 +1,637 @@
+"""
+This script compares two methods of detecting migrations:
+1. Migration program listener (listens Migration program) - detects markets via successful migration transactions
+2. Direct market account listener (listens Pump Fun AMM program aka PumpSwap) - detects markets via program account subscription
+
+The script tracks which method detects new markets first and provides detailed performance statistics.
+
+Note: multiple endpoints available. Scroll down to change providers which you want to test.
+"""
+
+import asyncio
+import base64
+import json
+import os
+import struct
+import time
+
+import aiohttp
+import base58
+import websockets
+from dotenv import load_dotenv
+from solders.pubkey import Pubkey
+
+load_dotenv()
+
+RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
+MIGRATION_PROGRAM_ID = Pubkey.from_string("39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg")
+PUMP_AMM_PROGRAM_ID = Pubkey.from_string("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA")
+QUOTE_MINT_SOL = base58.b58encode(bytes(Pubkey.from_string("So11111111111111111111111111111111111111112"))).decode()
+
+MARKET_DISCRIMINATOR = base58.b58encode(b'\xf1\x9am\x04\x11\xb1m\xbc').decode()
+MARKET_ACCOUNT_LENGTH = 8 + 1 + 2 + 32 * 6 + 8  # total size of known market structure
+
+
+class DetectionTracker:
+    """Tracks and analyzes detection times for both methods across providers"""
+    
+    def __init__(self):
+        self.migrations = {}  # {base_mint: {provider: timestamp}}
+        self.markets = {}     # {base_mint: {provider: timestamp}}
+        self.migration_messages = {}  # {provider: count}
+        self.market_messages = {}     # {provider: count}
+        self.start_time = time.time()
+        
+    def add_migration(self, base_mint, provider, timestamp):
+        """Record a migration detection event"""
+        if base_mint not in self.migrations:
+            self.migrations[base_mint] = {}
+        self.migrations[base_mint][provider] = timestamp
+        print(f"[MIGRATION] base_mint={base_mint} provider={provider} time={timestamp:.3f}")
+
+    def add_market(self, base_mint, provider, timestamp):
+        """Record a market detection event"""
+        if base_mint not in self.markets:
+            self.markets[base_mint] = {}
+        self.markets[base_mint][provider] = timestamp
+        print(f"[MARKET] base_mint={base_mint} provider={provider} time={timestamp:.3f}")
+    
+    def increment_migration_messages(self, provider):
+        """Count WebSocket messages received by migration listener"""
+        if provider not in self.migration_messages:
+            self.migration_messages[provider] = 0
+        self.migration_messages[provider] += 1
+        
+    def increment_market_messages(self, provider):
+        """Count WebSocket messages received by market listener"""
+        if provider not in self.market_messages:
+            self.market_messages[provider] = 0
+        self.market_messages[provider] += 1
+
+    def print_summary(self):
+        """Print detailed summary statistics of the comparison test"""
+        test_duration = time.time() - self.start_time
+        
+        # Count total messages
+        total_migration_messages = sum(self.migration_messages.values())
+        total_market_messages = sum(self.market_messages.values())
+        
+        print("\n=== Test Summary ===")
+        print(f"Test duration: {test_duration:.2f} seconds")
+        print(f"WebSocket messages received: {total_migration_messages + total_market_messages}")
+        print(f"  - Migration events: {total_migration_messages}")
+        print(f"  - Market events: {total_market_messages}")
+        
+        # Count unique tokens detected by each method
+        unique_migrations = set(self.migrations.keys())
+        unique_markets = set(self.markets.keys())
+        common_tokens = unique_migrations & unique_markets
+
+        print(f"Tokens detected: {len(unique_migrations | unique_markets)}")
+        print(f"  - Migration events: {len(unique_migrations)}")
+        print(f"  - Market events: {len(unique_markets)}")
+        print(f"  - Detected in both: {len(common_tokens)}\n")
+        
+        print("=== Provider Message Counts ===")
+        print("Provider                | Migration Messages | Market Messages | Total Messages")
+        print("-" * 80)
+        all_providers = set(self.migration_messages.keys()) | set(self.market_messages.keys())
+        for provider in sorted(all_providers):
+            migration_count = self.migration_messages.get(provider, 0)
+            market_count = self.market_messages.get(provider, 0)
+            total = migration_count + market_count
+            print(f"{provider:<22} | {migration_count:<18} | {market_count:<14} | {total}")
+        print()
+
+        print("=== Migration Event Provider Performance ===")
+        self._print_provider_performance(self.migrations)
+
+        print("\n=== Market Event Provider Performance ===")
+        self._print_provider_performance(self.markets)
+        
+        # Compare detection methods for tokens detected by both
+        if common_tokens:
+            print("\n=== Detection Timing Comparison: Migration vs Market ===")
+            print("Base Mint                                     | First Detection Method | First Provider | Time Delta (ms)")
+            print("-" * 100)
+            
+            migration_first = 0
+            market_first = 0
+            total_delta_ms = 0
+            
+            for base_mint in sorted(common_tokens):
+                # Find earliest time for each method
+                migration_time = min(self.migrations[base_mint].values()) if base_mint in self.migrations else float('inf')
+                market_time = min(self.markets[base_mint].values()) if base_mint in self.markets else float('inf')
+                
+                # Find provider with earliest time for each method
+                migration_provider = None
+                if base_mint in self.migrations:
+                    migration_provider = min(self.migrations[base_mint].items(), key=lambda x: x[1])[0]
+                
+                market_provider = None
+                if base_mint in self.markets:
+                    market_provider = min(self.markets[base_mint].items(), key=lambda x: x[1])[0]
+                
+                delta_ms = abs(migration_time - market_time) * 1000
+                total_delta_ms += delta_ms
+                
+                if migration_time < market_time:
+                    first_method = "Migration"
+                    first_provider = migration_provider
+                    migration_first += 1
+                else:
+                    first_method = "Market"
+                    first_provider = market_provider
+                    market_first += 1
+                    
+                print(f"{base_mint} | {first_method:<21} | {first_provider:<14} | {delta_ms:8.1f}")
+            
+            # Print statistics summary
+            if common_tokens:
+                avg_delta_ms = total_delta_ms / len(common_tokens)
+                print("\nSummary statistics:")
+                print(f"  - Migration detected first: {migration_first}/{len(common_tokens)} ({migration_first/len(common_tokens)*100:.1f}%)")
+                print(f"  - Market detected first: {market_first}/{len(common_tokens)} ({market_first/len(common_tokens)*100:.1f}%)")
+                print(f"  - Average timing difference: {avg_delta_ms:.1f} ms")
+
+    def _print_provider_performance(self, events_dict):
+        """Print performance metrics for providers using a specific detection method"""
+        # Count how many times each provider was first
+        first_count = {}
+        total_events = 0
+        
+        for base_mint, providers in events_dict.items():
+            total_events += 1
+            if not providers:
+                continue
+                
+            # Find the fastest provider for this event
+            fastest_provider = min(providers.items(), key=lambda x: x[1])[0]
+            if fastest_provider not in first_count:
+                first_count[fastest_provider] = 0
+            first_count[fastest_provider] += 1
+        
+        if not first_count:
+            print("No events detected")
+            return
+            
+        # Print rankings
+        print("Provider                | First Detections | Percentage")
+        print("-" * 60)
+        
+        for provider, count in sorted(first_count.items(), key=lambda x: x[1], reverse=True):
+            percentage = (count / total_events) * 100 if total_events > 0 else 0
+            print(f"{provider:<22} | {count:<16} | {percentage:.1f}%")
+            
+        # Calculate average latency between providers
+        self._print_provider_latency_matrix(events_dict)
+    
+    def _print_provider_latency_matrix(self, events_dict):
+        """Print a matrix of average latency between providers"""
+        # Get unique providers
+        all_providers = set()
+        for providers_data in events_dict.values():
+            all_providers.update(providers_data.keys())
+        
+        if len(all_providers) <= 1:
+            return
+            
+        providers_list = sorted(all_providers)
+        
+        print("\nAverage Latency Matrix (ms):")
+        # Print header
+        header = "           |"
+        for provider in providers_list:
+            header += f" {provider[:8]:>8} |"
+        print(header)
+        print("-" * len(header))
+        
+        # Calculate and print latency matrix
+        for provider1 in providers_list:
+            row = f"{provider1[:8]:>8} |"
+            for provider2 in providers_list:
+                if provider1 == provider2:
+                    row += "      — |"
+                    continue
+                    
+                # Calculate average latency
+                latencies = []
+                for base_mint, providers_data in events_dict.items():
+                    if provider1 in providers_data and provider2 in providers_data:
+                        latency_ms = (providers_data[provider2] - providers_data[provider1]) * 1000
+                        latencies.append(latency_ms)
+                
+                if latencies:
+                    avg_latency = sum(latencies) / len(latencies)
+                    row += f" {avg_latency:>+7.1f} |"
+                else:
+                    row += "      ? |"
+            print(row)
+
+
+# ============ MARKET DETECTION METHODS ============
+
+async def fetch_existing_market_pubkeys():
+    """
+    Fetch existing AMM market accounts from the blockchain
+    
+    Used to filter out already existing markets when detecting new ones
+    """
+    headers = {"Content-Type": "application/json"}
+    body = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "getProgramAccounts",
+        "params": [
+            str(PUMP_AMM_PROGRAM_ID),
+            {
+                "encoding": "base64",
+                "commitment": "processed",
+                "filters": [
+                    {"dataSize": MARKET_ACCOUNT_LENGTH},
+                    {"memcmp": {"offset": 0, "bytes": MARKET_DISCRIMINATOR}},
+                    {"memcmp": {"offset": 75, "bytes": QUOTE_MINT_SOL}}
+                ]
+            }
+        ]
+    }
+
+    async with aiohttp.ClientSession() as session:
+        async with session.post(RPC_ENDPOINT, headers=headers, json=body) as resp:
+            res = await resp.json()
+            return {account["pubkey"] for account in res.get("result", [])}
+
+
+def parse_market_account_data(data):
+    """
+    Parse binary market account data into a structured format
+    
+    This function matches the parser from the market listener script
+    """
+    parsed_data = {}
+    offset = 8  # Skip discriminator
+
+    fields = [
+        ("pool_bump", "u8"),
+        ("index", "u16"),
+        ("creator", "pubkey"),
+        ("base_mint", "pubkey"),
+        ("quote_mint", "pubkey"),
+        ("lp_mint", "pubkey"),
+        ("pool_base_token_account", "pubkey"),
+        ("pool_quote_token_account", "pubkey"),
+        ("lp_supply", "u64"),
+    ]
+
+    try:
+        for field_name, field_type in fields:
+            if field_type == "pubkey":
+                value = data[offset:offset + 32]
+                parsed_data[field_name] = base58.b58encode(value).decode("utf-8")
+                offset += 32
+            elif field_type in {"u64", "i64"}:
+                value = struct.unpack("<Q", data[offset:offset + 8])[0] if field_type == "u64" else struct.unpack("<q", data[offset:offset + 8])[0]
+                parsed_data[field_name] = value
+                offset += 8
+            elif field_type == "u16":
+                value = struct.unpack("<H", data[offset:offset + 2])[0]
+                parsed_data[field_name] = value
+                offset += 2
+            elif field_type == "u8":
+                value = data[offset]
+                parsed_data[field_name] = value
+                offset += 1
+    except Exception as e:
+        print(f"[ERROR] Failed to parse market data: {e}")
+
+    return parsed_data
+
+
+def parse_migrate_instruction(data):
+    """
+    Parse binary migration instruction data into a structured format
+    
+    This function matches the parser from the migration listener script
+    """
+    if len(data) < 8:
+        print(f"[ERROR] Data length too short: {len(data)} bytes")
+        return None
+
+    offset = 8  # Skip discriminator
+    parsed_data = {}
+
+    fields = [
+        ("timestamp", "i64"),
+        ("index", "u16"), 
+        ("creator", "pubkey"),
+        ("baseMint", "pubkey"),
+        ("quoteMint", "pubkey"),
+        ("baseMintDecimals", "u8"),
+        ("quoteMintDecimals", "u8"),
+        ("baseAmountIn", "u64"),
+        ("quoteAmountIn", "u64"),
+        ("poolBaseAmount", "u64"),
+        ("poolQuoteAmount", "u64"),
+        ("minimumLiquidity", "u64"),
+        ("initialLiquidity", "u64"),
+        ("lpTokenAmountOut", "u64"),
+        ("poolBump", "u8"),
+        ("pool", "pubkey"),
+        ("lpMint", "pubkey"),
+        ("userBaseTokenAccount", "pubkey"),
+        ("userQuoteTokenAccount", "pubkey"),
+    ]
+
+    try:
+        for field_name, field_type in fields:
+            if field_type == "pubkey":
+                value = data[offset:offset + 32]
+                parsed_data[field_name] = base58.b58encode(value).decode("utf-8")
+                offset += 32
+            elif field_type in {"u64", "i64"}:
+                value = struct.unpack("<Q", data[offset:offset + 8])[0] if field_type == "u64" else struct.unpack("<q", data[offset:offset + 8])[0]
+                parsed_data[field_name] = value
+                offset += 8
+            elif field_type == "u16":
+                value = struct.unpack("<H", data[offset:offset + 2])[0]
+                parsed_data[field_name] = value
+                offset += 2
+            elif field_type == "u8":
+                value = data[offset]
+                parsed_data[field_name] = value
+                offset += 1
+
+        return parsed_data
+
+    except Exception as e:
+        print(f"[ERROR] Failed to parse migration data at offset {offset}: {e}")
+        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 ============
+
+async def listen_for_migrations(wss_url, provider_name, tracker, known_events=None):
+    """
+    Listen for migration instructions via WebSocket
+    
+    Args:
+        wss_url: WebSocket URL to connect to
+        provider_name: Name of the RPC provider
+        tracker: DetectionTracker instance to record events
+        known_events: Set of already known (provider, base_mint) tuples to prevent duplicates
+    """
+    if known_events is None:
+        known_events = set()
+        
+    while True:
+        try:
+            print(f"[INFO] Connecting migration listener to {provider_name}...")
+            async with websockets.connect(wss_url) as websocket:
+                # Subscribe to logs mentioning the migration program
+                subscription_message = json.dumps({
+                    "jsonrpc": "2.0",
+                    "id": 1,
+                    "method": "logsSubscribe",
+                    "params": [
+                        {"mentions": [str(MIGRATION_PROGRAM_ID)]},
+                        {"commitment": "processed"},
+                    ],
+                })
+                await websocket.send(subscription_message)
+                await websocket.recv()  # Get subscription confirmation
+                print(f"[INFO] Migration listener active for {provider_name}")
+
+                while True:
+                    try:
+                        # Receive WebSocket message
+                        response = await websocket.recv()
+                        data = json.loads(response)
+                        tracker.increment_migration_messages(provider_name)
+                        
+                        # Check if it's a notification and not something else
+                        if data.get("method") != "logsNotification":
+                            continue
+                            
+                        # Get transaction logs
+                        log_data = data["params"]["result"]["value"]
+                        logs = log_data.get("logs", [])
+                        
+                        # Skip failed transactions
+                        if not is_transaction_successful(logs):
+                            continue
+                            
+                        # Skip if not a Migrate instruction
+                        if not any("Instruction: Migrate" in log for log in logs):
+                            continue
+                            
+                        # Skip already migrated curves
+                        if any("already migrated" in log for log in logs):
+                            continue
+                            
+                        # Search for Program data in logs
+                        for log in logs:
+                            if log.startswith("Program data:"):
+                                try:
+                                    # Decode and parse the instruction data
+                                    data = base64.b64decode(log.split(": ")[1])
+                                    parsed = parse_migrate_instruction(data)
+                                    
+                                    if parsed and "baseMint" in parsed:
+                                        base_mint = parsed["baseMint"]
+                                        # Only track the timestamp for the first time we see this event
+                                        # from this provider, but still count messages
+                                        if (provider_name, base_mint) not in known_events:
+                                            ts = time.time()
+                                            tracker.add_migration(base_mint, provider_name, ts)
+                                            known_events.add((provider_name, base_mint))
+                                    break
+                                except Exception as e:
+                                    print(f"[ERROR] Failed to decode Program data: {e}")
+                                
+                    except Exception as e:
+                        print(f"[ERROR] Migration listener for {provider_name}: {e}")
+
+        except Exception as e:
+            print(f"[ERROR] Connection error in migration listener for {provider_name}: {e}")
+            print("[INFO] Reconnecting in 5 seconds...")
+            await asyncio.sleep(5)
+
+
+async def listen_for_markets(wss_url, provider_name, tracker, known_markets):
+    """
+    Listen for new market accounts via WebSocket
+    
+    Args:
+        wss_url: WebSocket URL to connect to
+        provider_name: Name of the RPC provider
+        tracker: DetectionTracker instance to record events
+        known_markets: Set of already known market pubkeys to prevent duplicates
+    """
+    while True:
+        try:
+            print(f"[INFO] Connecting market listener to {provider_name}...")
+            async with websockets.connect(wss_url) as websocket:
+                # Subscribe to program account changes
+                sub_msg = json.dumps({
+                    "jsonrpc": "2.0",
+                    "id": 1,
+                    "method": "programSubscribe",
+                    "params": [
+                        str(PUMP_AMM_PROGRAM_ID),
+                        {
+                            "commitment": "processed",
+                            "encoding": "base64",
+                            "filters": [
+                                {"dataSize": MARKET_ACCOUNT_LENGTH},
+                                {"memcmp": {"offset": 0, "bytes": MARKET_DISCRIMINATOR}},
+                                {"memcmp": {"offset": 75, "bytes": QUOTE_MINT_SOL}}
+                            ]
+                        }
+                    ]
+                })
+                await websocket.send(sub_msg)
+                await websocket.recv()  # Get subscription confirmation
+                print(f"[INFO] Market listener active for {provider_name}")
+                
+                # Track events already seen by this provider
+                provider_known = set()
+                
+                while True:
+                    try:
+                        # Receive WebSocket message
+                        msg = await websocket.recv()
+                        data = json.loads(msg)
+                        tracker.increment_market_messages(provider_name)
+                        
+                        # Check if it's a notification and not something else
+                        if data.get("method") != "programNotification":
+                            continue
+                            
+                        # Extract account information
+                        message_value = data["params"]["result"]["value"]
+                        pubkey = message_value["pubkey"]
+                        raw_account_data = message_value["account"].get("data", [None])[0]
+                        
+                        # Skip if we've already processed this market (either globally or for this provider)
+                        if pubkey in known_markets or pubkey in provider_known:
+                            continue
+                        provider_known.add(pubkey)
+                        
+                        # Skip if there's no data
+                        if not raw_account_data:
+                            print("[ERROR] Account data is empty")
+                            continue
+
+                        try:
+                            # Decode and parse the account data
+                            account_data = base64.b64decode(raw_account_data)
+                            parsed = parse_market_account_data(account_data)
+
+                            # Skip user-created markets (they are on-curve)
+                            if parsed.get("creator") and Pubkey.from_string(parsed.get("creator")).is_on_curve():
+                                continue  # skip user-created pool
+
+                            # Record the market detection
+                            base_mint = parsed.get("base_mint")
+                            if base_mint:
+                                ts = time.time()
+                                tracker.add_market(base_mint, provider_name, ts)
+                                
+                                # Add to the shared known markets to avoid duplicate processing
+                                known_markets.add(pubkey)
+                                
+                        except Exception as e:
+                            print(f"[ERROR] Failed to decode account: {e}")
+
+                    except Exception as e:
+                        print(f"[ERROR] Market listener for {provider_name}: {e}")
+
+        except Exception as e:
+            print(f"[ERROR] Connection error in market listener for {provider_name}: {e}")
+            print("[INFO] Reconnecting in 5 seconds...")
+            await asyncio.sleep(5)
+
+
+# ============ MAIN TEST RUNNER ============
+
+async def run_comparison_test(migration_wss_endpoints, market_wss_endpoints, test_duration=600):
+    """
+    Run the comparison test with multiple WebSocket endpoints
+    
+    Args:
+        migration_wss_endpoints: Dict of {provider_name: wss_url} for migration listeners
+        market_wss_endpoints: Dict of {provider_name: wss_url} for market listeners
+        test_duration: How long to run the test in seconds (default: 10 minutes)
+    """
+    # Initialize our tracker and fetch existing markets to avoid duplicates
+    tracker = DetectionTracker()
+    known_markets = await fetch_existing_market_pubkeys()
+    print(f"[INFO] Loaded {len(known_markets)} existing markets")
+
+    known_migration_events = set()    
+    tasks = []
+    
+    # Start migration listeners
+    for provider_name, wss_url in migration_wss_endpoints.items():
+        print(f"[INFO] Starting migration listener for {provider_name}")
+        task = asyncio.create_task(
+            listen_for_migrations(wss_url, provider_name, tracker, known_migration_events)
+        )
+        tasks.append(task)
+    
+    # Start market listeners
+    for provider_name, wss_url in market_wss_endpoints.items():
+        print(f"[INFO] Starting market listener for {provider_name}")
+        task = asyncio.create_task(
+            listen_for_markets(wss_url, provider_name, tracker, known_markets)
+        )
+        tasks.append(task)
+    
+    # Run for specified duration
+    print(f"[INFO] Test running for {test_duration} seconds...")
+    await asyncio.sleep(test_duration)
+
+    for task in tasks:
+        task.cancel()
+    
+    await asyncio.gather(*tasks, return_exceptions=True)
+    tracker.print_summary()
+
+
+if __name__ == "__main__":
+    # Read providers from environment variables
+    # You can add more providers by adding additional environment variables
+    migration_providers = {
+        "chainstack": os.environ.get("SOLANA_NODE_WSS_ENDPOINT"),
+        "provider_2": os.environ.get("SOLANA_NODE_WSS_ENDPOINT_2"),
+        # Add more providers to .env as needed
+    }
+    
+    market_providers = {
+        "chainstack": os.environ.get("SOLANA_NODE_WSS_ENDPOINT"),
+        "provider_2": os.environ.get("SOLANA_NODE_WSS_ENDPOINT_2"),
+        # Add more providers to .env as needed
+    }
+    
+    # Filter out any providers with missing endpoints
+    migration_providers = {name: url for name, url in migration_providers.items() if url}
+    market_providers = {name: url for name, url in market_providers.items() if url}
+    
+    # Get test duration from environment or use default (10 minutes)
+    TEST_DURATION = int(os.environ.get("TEST_DURATION", 600))
+    
+    print(f"[INFO] Starting Solana detector comparison test for {TEST_DURATION} seconds")
+    print(f"[INFO] Migration providers: {', '.join(migration_providers.keys())}")
+    print(f"[INFO] Market providers: {', '.join(market_providers.keys())}")
+    
+    # Run the test
+    asyncio.run(run_comparison_test(migration_providers, market_providers, test_duration=TEST_DURATION))
@@ -1,15 +1,15 @@
 import asyncio
 import json
 import os
-import sys

 import websockets
+from dotenv import load_dotenv
+from solders.pubkey import Pubkey

-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-
-from core.pubkeys import PumpAddresses
+load_dotenv()

 WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
+PUMP_MIGRATOR_ID = Pubkey.from_string("39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg")


 def process_initialize2_transaction(data):
@@ -48,7 +48,7 @@ async def listen_for_events():
                        "params": [
                            {
                                "mentionsAccountOrProgram": str(
-                                    PumpAddresses.LIQUIDITY_MIGRATOR
+                                    PUMP_MIGRATOR_ID
                                )
                            },
                            {
@@ -1,3 +1,11 @@
+"""
+Listens for 'Migrate' instructions from a Solana migration program via WebSocket.
+Parses and logs transaction details (e.g., mint, liquidity, token accounts) for successful migrations.
+
+Note: skips transactions with truncated logs (no Program data in the logs -> no parsed data).
+To cover those cases, please use an additional RPC call (get transaction data) or additional listener not based on logs.
+"""
+
 import asyncio
 import base64
 import json
@@ -6,8 +14,11 @@ import struct

 import base58
 import websockets
+from dotenv import load_dotenv
 from solders.pubkey import Pubkey

+load_dotenv()
+
 WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
 MIGRATION_PROGRAM_ID = Pubkey.from_string("39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg")

@@ -77,9 +88,8 @@ def is_transaction_successful(logs):


 def print_transaction_details(log_data):
-    signature = log_data.get('signature', 'N/A')
-    print(f"\n[INFO] Transaction Signature: {signature}")
    logs = log_data.get("logs", [])
+    parsed_data = {}

    for log in logs:
        if log.startswith("Program data:"):
@@ -87,16 +97,14 @@ def print_transaction_details(log_data):
                data = base64.b64decode(log.split(": ")[1])
                parsed_data = parse_migrate_instruction(data)
                if parsed_data:
-                    print("\n[INFO] Parsed Migration Instruction Data:")
+                    print("[INFO] Parsed from Program data:")
                    for key, value in parsed_data.items():
                        print(f"  {key}: {value}")
            except Exception as e:
-                print(f"[ERROR] Base64 decode failed: {log}, Error: {e}")
+                print(f"[ERROR] Failed to decode Program data: {e}")

-def print_log_details(logs):
-    print("\n[INFO] Processing logs:")
-    for log in logs:
-        print(f"  Log: {log}")
+    if not parsed_data:
+        print("[ERROR] Failed to parse migration data: parsed data is empty")


 async def listen_for_migrations():
@@ -130,16 +138,19 @@ async def listen_for_migrations():
                            log_data = data["params"]["result"]["value"]
                            logs = log_data.get("logs", [])

+                            signature = log_data.get('signature', 'N/A')
+                            print(f"\n[INFO] Transaction signature: {signature}")
+
                            if is_transaction_successful(logs):
                                if not any("Program log: Instruction: Migrate" in log for log in logs):
-                                    print("[INFO] Skipping: No Migrate instruction")
+                                    print("[INFO] Skipping: no migrate instruction")
                                    continue

                                if any("Program log: Bonding curve already migrated" in log for log in logs):
-                                    print("[INFO] Skipping: Bonding curve already migrated")
+                                    print("[INFO] Skipping: bonding curve already migrated")
                                    continue

-                                print("[INFO] Processing migration instruction!")
+                                print("[INFO] Processing migration instruction...")
                                print_transaction_details(log_data)
                            else:
                                print("[INFO] Skipping failed transaction.")
@@ -155,4 +166,4 @@ async def listen_for_migrations():
            await asyncio.sleep(5)

 if __name__ == "__main__":
-    asyncio.run(listen_for_migrations())
+    asyncio.run(listen_for_migrations())
@@ -0,0 +1,169 @@
+"""
+Monitors Solana for new Pump AMM markets via WebSocket.
+Fetches existing markets to filter out already existing ones, parses market account data (e.g., mints, token accounts, creator),
+and excludes user-created markets. May also detect non-migration-based markets (if they created by a program).
+
+Note: this method consumes HUGE AMOUNT OF MESSAGES from a WebSocket.
+"""
+
+import asyncio
+import base64
+import json
+import os
+import struct
+
+import aiohttp
+import base58
+import websockets
+from dotenv import load_dotenv
+from solders.pubkey import Pubkey
+
+load_dotenv()
+
+WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
+RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT")
+PUMP_AMM_PROGRAM_ID = Pubkey.from_string("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA")
+
+MARKET_ACCOUNT_LENGTH = 8 + 1 + 2 + 32 * 6 + 8  # total size of known market structure
+MARKET_DISCRIMINATOR = base58.b58encode(b'\xf1\x9am\x04\x11\xb1m\xbc').decode()
+QUOTE_MINT_SOL = base58.b58encode(bytes(Pubkey.from_string("So11111111111111111111111111111111111111112"))).decode()  
+
+
+async def fetch_existing_market_pubkeys():
+    headers = {"Content-Type": "application/json"}
+    body = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "getProgramAccounts",
+        "params": [
+            str(PUMP_AMM_PROGRAM_ID),
+            {
+                "encoding": "base64",
+                "commitment": "processed",
+                "filters": [
+                    {"dataSize": MARKET_ACCOUNT_LENGTH},
+                    {"memcmp": {"offset": 0, "bytes": MARKET_DISCRIMINATOR}},
+                    {"memcmp": {"offset": 75, "bytes": QUOTE_MINT_SOL}}
+                ]
+            }
+        ]
+    }
+
+    async with aiohttp.ClientSession() as session:
+        async with session.post(RPC_ENDPOINT, headers=headers, json=body) as resp:
+            res = await resp.json()
+            return {account["pubkey"] for account in res.get("result", [])}
+
+
+def parse_market_account_data(data):
+    parsed_data = {}
+    offset = 8  # Discriminator
+
+    fields = [
+        ("pool_bump", "u8"),
+        ("index", "u16"),
+        ("creator", "pubkey"),
+        ("base_mint", "pubkey"),
+        ("quote_mint", "pubkey"),
+        ("lp_mint", "pubkey"),
+        ("pool_base_token_account", "pubkey"),
+        ("pool_quote_token_account", "pubkey"),
+        ("lp_supply", "u64"),
+    ]
+
+    try:
+        for field_name, field_type in fields:
+            if field_type == "pubkey":
+                value = data[offset:offset + 32]
+                parsed_data[field_name] = base58.b58encode(value).decode("utf-8")
+                offset += 32
+            elif field_type in {"u64", "i64"}:
+                value = struct.unpack("<Q", data[offset:offset + 8])[0] if field_type == "u64" else struct.unpack("<q", data[offset:offset + 8])[0]
+                parsed_data[field_name] = value
+                offset += 8
+            elif field_type == "u16":
+                value = struct.unpack("<H", data[offset:offset + 2])[0]
+                parsed_data[field_name] = value
+                offset += 2
+            elif field_type == "u8":
+                value = data[offset]
+                parsed_data[field_name] = value
+                offset += 1
+    except Exception as e:
+        print(f"[ERROR] Failed to parse market data: {e}")
+
+    return parsed_data
+
+
+async def listen_new_markets():
+    known_pubkeys = await fetch_existing_market_pubkeys()
+    print(f"[INFO] Loaded {len(known_pubkeys)} existing markets")
+
+    while True:
+        try:
+            print("[INFO] Connecting to WebSocket...")
+            async with websockets.connect(WSS_ENDPOINT) as ws:
+                sub_msg = json.dumps({
+                    "jsonrpc": "2.0",
+                    "id": 1,
+                    "method": "programSubscribe",
+                    "params": [
+                        str(PUMP_AMM_PROGRAM_ID),
+                        {
+                            "commitment": "processed",
+                            "encoding": "base64",
+                            "filters": [
+                                {"dataSize": MARKET_ACCOUNT_LENGTH},
+                                {"memcmp": {"offset": 0, "bytes": MARKET_DISCRIMINATOR}},
+                                {"memcmp": {"offset": 75, "bytes": QUOTE_MINT_SOL}}
+                            ]
+                        }
+                    ]
+                })
+                await ws.send(sub_msg)
+                print(f"[INFO] Subscribed to: {PUMP_AMM_PROGRAM_ID}")
+
+                while True:
+                    msg = await ws.recv()
+                    data = json.loads(msg)
+
+                    if "method" in data and data["method"] == "programNotification":
+                        message_value = data["params"]["result"]["value"]
+                        pubkey = message_value["pubkey"]
+                        raw_account_data = message_value["account"].get("data", [None])[0]
+                        slot = data["params"]["result"]["context"]["slot"]
+
+                        if pubkey in known_pubkeys:
+                            #print("[INFO] Skipping already existed market...")
+                            continue
+
+                        if not raw_account_data:
+                            print("[ERROR] Account data is empty")
+                            continue
+
+                        try:
+                            account_data = base64.b64decode(raw_account_data)
+                            parsed = parse_market_account_data(account_data)
+
+                            if Pubkey.from_string(parsed.get("creator", "")).is_on_curve():
+                                print("[INFO] Skipping user-created market...")
+                                continue  # skip user-created pool
+
+                            print("\n[INFO] New market account detected:")
+                            print(f"  pubkey: {pubkey}")
+                            print(f"  slot: {slot}")
+                            for k, v in parsed.items():
+                                print(f"  {k}: {v}")
+
+                            known_pubkeys.add(pubkey)
+                        except Exception as e:
+                            print(f"[ERROR] Failed to decode account: {e}")
+
+        except Exception as e:
+            print(f"[ERROR] Connection error: {e}")
+            print("[INFO] Reconnecting in 5 seconds...")
+            await asyncio.sleep(5)
+
+
+if __name__ == "__main__":
+    asyncio.run(listen_new_markets())
@@ -1,17 +1,25 @@
+"""
+Listens to Solana blocks for Pump.fun 'create' instructions via WebSocket.
+Decodes transaction data to extract mint, bonding curve, and user details.
+
+It is usually slower than other listeners.
+"""
+
 import asyncio
 import base64
 import json
 import os
 import struct
-import sys

 import websockets
+from dotenv import load_dotenv
+from solders.pubkey import Pubkey
 from solders.transaction import VersionedTransaction

-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from core.pubkeys import PumpAddresses
+load_dotenv()

 WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
+PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")


 def load_idl(file_path):
@@ -58,7 +66,7 @@ async def listen_and_decode_create():
                "id": 1,
                "method": "blockSubscribe",
                "params": [
-                    {"mentionsAccountOrProgram": str(PumpAddresses.PROGRAM)},
+                    {"mentionsAccountOrProgram": str(PUMP_PROGRAM_ID)},
                    {
                        "commitment": "confirmed",
                        "encoding": "base64",
@@ -70,7 +78,7 @@ async def listen_and_decode_create():
            }
        )
        await websocket.send(subscription_message)
-        print(f"Subscribed to blocks mentioning program: {PumpAddresses.PROGRAM}")
+        print(f"Subscribed to blocks mentioning program: {PUMP_PROGRAM_ID}")

        while True:
            try:
@@ -97,7 +105,7 @@ async def listen_and_decode_create():
                                                transaction.message.account_keys[
                                                    ix.program_id_index
                                                ]
-                                            ) == str(PumpAddresses.PROGRAM):
+                                            ) == str(PUMP_PROGRAM_ID):
                                                ix_data = bytes(ix.data)
                                                discriminator = struct.unpack(
                                                    "<Q", ix_data[:8]
@@ -1,3 +1,11 @@
+"""
+Monitors Solana for new Pump.fun token creations using Geyser gRPC.
+Decodes 'create' instructions to extract and display token details (name, symbol, mint, bonding curve).
+Requires a Geyser API token for access.
+
+It is proven to be the fastest listener.
+"""
+
 import asyncio
 import os
 import struct
@@ -6,6 +14,7 @@ import base58
 import grpc
 from dotenv import load_dotenv
 from generated import geyser_pb2, geyser_pb2_grpc
+from solders.pubkey import Pubkey

 load_dotenv()

@@ -13,14 +22,14 @@ load_dotenv()
 GEYSER_ENDPOINT = os.getenv("GEYSER_ENDPOINT")
 GEYSER_API_TOKEN = os.getenv("GEYSER_API_TOKEN")

-PUMP_PROGRAM = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
+PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
 PUMP_CREATE_PREFIX = struct.pack("<Q", 8576854823835016728)


 async def create_geyser_connection():
    """Establish a secure connection to the Geyser endpoint."""
    auth = grpc.metadata_call_credentials(
-        lambda context, callback: callback((("authorization", f"Basic {GEYSER_API_TOKEN}"),), None)
+        lambda _, callback: callback((("authorization", f"Basic {GEYSER_API_TOKEN}"),), None)
    )
    creds = grpc.composite_channel_credentials(grpc.ssl_channel_credentials(), auth)
    channel = grpc.aio.secure_channel(GEYSER_ENDPOINT, creds)
@@ -30,7 +39,7 @@ async def create_geyser_connection():
 def create_subscription_request():
    """Create a subscription request for Pump.fun transactions."""
    request = geyser_pb2.SubscribeRequest()
-    request.transactions["pump_filter"].account_include.append(PUMP_PROGRAM)
+    request.transactions["pump_filter"].account_include.append(str(PUMP_PROGRAM_ID))
    request.transactions["pump_filter"].failed = False
    request.commitment = geyser_pb2.CommitmentLevel.PROCESSED
    return request
@@ -1,19 +1,28 @@
+"""
+Listens for new Pump.fun token creations via Solana WebSocket.
+Monitors logs for 'Create' instructions, decodes and prints token details (name, symbol, mint, etc.).
+Additionally, calculates an associated bonding curve address for each token.
+
+It is usually faster than blockSubscribe, but slower than Geyser.
+"""
+
 import asyncio
 import base64
 import json
 import os
 import struct
-import sys

 import base58
 import websockets
+from dotenv import load_dotenv
 from solders.pubkey import Pubkey

-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from core.pubkeys import PumpAddresses, SystemAddresses
+load_dotenv()

 WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
-
+PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
+TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
+ASSOCIATED_TOKEN_PROGRAM_ID = Pubkey.from_string("ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL")

 def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey:
    """
@@ -23,24 +32,14 @@ def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey
    derived_address, _ = Pubkey.find_program_address(
        [
            bytes(bonding_curve),
-            bytes(SystemAddresses.TOKEN_PROGRAM),
+            bytes(TOKEN_PROGRAM_ID),
            bytes(mint),
        ],
-        SystemAddresses.ASSOCIATED_TOKEN_PROGRAM,
+        ASSOCIATED_TOKEN_PROGRAM_ID,
    )
    return derived_address


-# Load the IDL JSON file
-with open("idl/pump_fun_idl.json") as f:
-    idl = json.load(f)
-
-# Extract the "create" instruction definition
-create_instruction = next(
-    instr for instr in idl["instructions"] if instr["name"] == "create"
-)
-
-
 def parse_create_instruction(data):
    if len(data) < 8:
        return None
@@ -75,18 +74,6 @@ def parse_create_instruction(data):
        return None


-def print_transaction_details(log_data):
-    print(f"Signature: {log_data.get('signature')}")
-
-    for log in log_data.get("logs", []):
-        if log.startswith("Program data:"):
-            try:
-                data = base58.b58decode(log.split(": ")[1]).decode("utf-8")
-                print(f"Data: {data}")
-            except:
-                pass
-
-
 async def listen_for_new_tokens():
    while True:
        try:
@@ -97,14 +84,14 @@ async def listen_for_new_tokens():
                        "id": 1,
                        "method": "logsSubscribe",
                        "params": [
-                            {"mentions": [str(PumpAddresses.PROGRAM)]},
+                            {"mentions": [str(PUMP_PROGRAM_ID)]},
                            {"commitment": "processed"},
                        ],
                    }
                )
                await websocket.send(subscription_message)
                print(
-                    f"Listening for new token creations from program: {PumpAddresses.PROGRAM}"
+                    f"Listening for new token creations from program: {PUMP_PROGRAM_ID}"
                )

                # Wait for subscription confirmation
@@ -1,27 +1,25 @@
+"""
+Listens for new Pump.fun token creations via Solana WebSocket.
+Monitors logs for 'Create' instructions, decodes and prints token details (name, symbol, mint, etc.).
+
+It is usually faster than blockSubscribe, but slower than Geyser.
+"""
+
 import asyncio
 import base64
 import json
 import os
 import struct
-import sys

 import base58
 import websockets
+from dotenv import load_dotenv
+from solders.pubkey import Pubkey

-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from core.pubkeys import PumpAddresses
+load_dotenv()

 WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
-
-
-# Load the IDL JSON file
-with open("idl/pump_fun_idl.json") as f:
-    idl = json.load(f)
-
-# Extract the "create" instruction definition
-create_instruction = next(
-    instr for instr in idl["instructions"] if instr["name"] == "create"
-)
+PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")


 def parse_create_instruction(data):
@@ -58,18 +56,6 @@ def parse_create_instruction(data):
        return None


-def print_transaction_details(log_data):
-    print(f"Signature: {log_data.get('signature')}")
-
-    for log in log_data.get("logs", []):
-        if log.startswith("Program data:"):
-            try:
-                data = base58.b58decode(log.split(": ")[1]).decode("utf-8")
-                print(f"Data: {data}")
-            except:
-                pass
-
-
 async def listen_for_new_tokens():
    while True:
        try:
@@ -80,14 +66,14 @@ async def listen_for_new_tokens():
                        "id": 1,
                        "method": "logsSubscribe",
                        "params": [
-                            {"mentions": [str(PumpAddresses.PROGRAM)]},
+                            {"mentions": [str(PUMP_PROGRAM_ID)]},
                            {"commitment": "processed"},
                        ],
                    }
                )
                await websocket.send(subscription_message)
                print(
-                    f"Listening for new token creations from program: {PumpAddresses.PROGRAM}"
+                    f"Listening for new token creations from program: {PUMP_PROGRAM_ID}"
                )

                # Wait for subscription confirmation
@@ -1,3 +1,7 @@
+"""
+Listens for new Pump.fun token creations via PumpPortal WebSocket.
+"""
+
 import asyncio
 import json
 from datetime import datetime
@@ -1,3 +1,9 @@
+"""
+This module provides functionality to:
+1. Find market addresses by base mint
+2. Fetch and parse market data (including pool addresses) from Pump AMM program accounts
+"""
+
 import asyncio
 import os
 import struct
@@ -1,3 +1,12 @@
+"""
+This module provides functionality to:
+- Find market addresses by token mint.
+- Fetch and parse market data from PUMP AMM pools.
+- Calculate token prices in AMM pools.
+- Create associated token accounts (ATAs) idempotently.
+- Sell tokens on the PUMP AMM with slippage protection.
+"""
+
 import asyncio
 import os
 import struct
@@ -1,96 +0,0 @@
-"""
-Track bonding curve progress for a pump.fun token by mint address.
-"""
-
-import asyncio
-import os
-import struct
-
-from dotenv import load_dotenv
-from solana.rpc.async_api import AsyncClient
-from solders.pubkey import Pubkey
-
-# Import pump.fun program address
-from core.pubkeys import PumpAddresses
-
-load_dotenv()
-
-RPC_URL = os.getenv("SOLANA_NODE_RPC_ENDPOINT")
-TOKEN_MINT = "xWrzYY4c1LnbSkLrd2LDUg9vw7YtVyJhGmw7MABpump"
-
-LAMPORTS_PER_SOL = 1_000_000_000
-TOKEN_DECIMALS = 6
-EXPECTED_DISCRIMINATOR = struct.pack("<Q", 6966180631402821399)  # pump.fun bonding curve discriminator
-POLL_INTERVAL = 5  # seconds between each status check
-
-
-def get_associated_bonding_curve_address(mint: Pubkey, program_id: Pubkey) -> Pubkey:
-    """Derive the bonding curve PDA address from mint."""
-    return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)[0]
-
-
-async def get_account_data(client: AsyncClient, pubkey: Pubkey) -> bytes:
-    """Fetch raw account data for a given public key."""
-    resp = await client.get_account_info(pubkey, encoding="base64")
-    if not resp.value or not resp.value.data:
-        raise ValueError(f"Account {pubkey} not found or has no data")
-
-    return resp.value.data
-
-
-def parse_curve_state(data: bytes) -> dict:
-    """Decode bonding curve account data."""
-    if data[:8] != EXPECTED_DISCRIMINATOR:
-        raise ValueError("Invalid discriminator for bonding curve")
-
-    fields = struct.unpack_from("<QQQQQ?", data, 8)
-    return {
-        "virtual_token_reserves": fields[0] / 10**TOKEN_DECIMALS,
-        "virtual_sol_reserves": fields[1] / LAMPORTS_PER_SOL,
-        "real_token_reserves": fields[2] / 10**TOKEN_DECIMALS,
-        "real_sol_reserves": fields[3] / LAMPORTS_PER_SOL,
-        "token_total_supply": fields[4] / 10**TOKEN_DECIMALS,
-        "complete": fields[5],
-    }
-
-
-def print_curve_status(state: dict):
-    """Print current bonding curve status."""
-    progress = 0
-    if state["token_total_supply"]:
-        progress = 100 - (100 * state["real_token_reserves"] / state["token_total_supply"])
-
-    print("==============================")
-    print(f"Complete: {'✅' if state['complete'] else '❌'}")
-    print(f"Progress: {progress:.2f}%")
-    print(f"Token Reserves: {state['real_token_reserves']:.4f}")
-    print(f"SOL Reserves:   {state['real_sol_reserves']:.4f}")
-    print("==============================\n")
-
-
-async def track_curve():
-    """Main loop to track bonding curve state."""
-    if not RPC_URL or not TOKEN_MINT:
-        print("❌ Set SOLANA_NODE_RPC_ENDPOINT and TOKEN_MINT in .env")
-        return
-
-    mint_pubkey = Pubkey.from_string(TOKEN_MINT)
-    curve_pubkey = get_associated_bonding_curve_address(mint_pubkey, PumpAddresses.PROGRAM)
-
-    print("Tracking bonding curve for:", mint_pubkey)
-    print("Curve address:", curve_pubkey, "\n")
-
-    async with AsyncClient(RPC_URL) as client:
-        while True:
-            try:
-                data = await get_account_data(client, curve_pubkey)
-                state = parse_curve_state(data)
-                print_curve_status(state)
-            except Exception as e:
-                print(f"⚠️ Error: {e}")
-
-            await asyncio.sleep(POLL_INTERVAL)
-
-
-if __name__ == "__main__":
-    asyncio.run(track_curve())