Feature for checking for snipers and bundles.

[akiraloli]

Feature request description by @slimeonmyhead

Create a tool to check for bundles/snipers for a given mint address on solana.
This checker should catch snipers that originate outside the original mint bundle from pumpfun.
Each bundle identified should contain stats such as %supply, sol spent, # of unique wallets, how much they are still holding, etc.
Include an overall summary of the bundles

OG Request: It would be cool to be able to check if a token is bundled directly from neur. Adding on to this, I think more features around being able to do a full "due diligence" analysis directly from neur should be the goal. I can't think of anything right now but I'll request more when i do.
For now checking for bundles, snipes, bubble maps, mint auth, freeze auth, LP burn, and more of that nature are good features to be able to check directly from neur. Thanks :)

[Tejoyasaswi-17]

@backwardsmap I would like to pick this issue. But on what criteria will we flag snipers. With the help of bitquery api's we can get the pump fun data?

[briang123]

I asked ChatGPT what it thought.

To build a tool that analyzes bundles, snipers, and other attributes for a given mint address on Solana, we need a clear structure and defined criteria for flagging suspicious activity. Below is a breakdown of how we can design this tool, criteria for flagging snipers, and implementation suggestions using APIs like Bitquery or other Solana blockchain explorers.

Tool Features
1. Bundle Analysis:
• Identify bundles originating from pump.fun or other verified sources.
• Provide stats for each bundle:
• % of Supply: Percentage of total token supply within the bundle.
• SOL Spent: Total amount spent on the mint.
• Unique Wallets: Number of distinct wallets involved in the bundle.
• Holdings: Total tokens still held by the wallets in the bundle.
2. Snipe Detection:
• Identify tokens acquired outside the original pump.fun mint bundle.
• Flag wallets that acquired tokens disproportionately early (e.g., within seconds of the mint).
• Analyze:
• Transaction Speed: Wallets executing transactions faster than typical minting speed.
• Wallet Behavior: High-value purchases immediately dumped into the market.
• Anomalous Patterns: Bulk buys without participating in the mint.
3. Bubble Maps:
• Visualize token transfers between wallets to detect clusters or coordinated activity.
• Highlight wallets controlling significant portions of supply.
4. Mint & Freeze Authority:
• Check and validate:
• Mint Authority: Who has the ability to mint additional tokens.
• Freeze Authority: Who can freeze token transfers.
5. Liquidity Pool (LP) Burn Analysis:
• Monitor LP creation and ensure tokens added to pools match the project’s claims.
• Detect LP tokens that are being burned or removed in suspicious amounts.

Sniper Criteria
1. Speed-Based:
• Transactions executed within milliseconds of token mint availability.
• Use timestamp data to compare mint transactions vs. sniper transactions.
2. Quantity-Based:
• Acquiring large amounts of supply (e.g., >1% or 2%) within a short window.
• Disproportionate holdings compared to average wallet holdings.
3. Wallet History:
• Wallets with previous sniper activity in other token launches.
• Wallets showing patterns of quick buy-sell actions to exploit initial hype.
4. Transaction Sources:
• Identify purchases coming directly from DEXs or secondary markets instead of minting.

Implementation Steps
1. Use APIs to Extract Blockchain Data:
• Bitquery or Solana RPC APIs:
• Pull data for token mint transactions.
• Analyze token distribution and wallet behaviors.
• Use Pump.fun APIs to validate the original bundle transactions.
2. Data Analysis and Aggregation:
• Store mint and transfer data in a database.
• Group transactions by wallet and time to identify bundles and snipers.
3. Visualizations:
• Integrate bubble maps using tools like D3.js to visualize wallet clusters.
• Display key stats (e.g., % supply held, SOL spent, etc.) in an intuitive UI.
4. Summary Reporting:
• Generate an overview for the mint:
• % Supply in bundles.
• Number of snipers flagged.
• Largest holders and their behavior.
5. Future Features:
• Add “due diligence” capabilities like LP health, tokenomics verification, etc.
• Allow on-demand checking of mint authorities, LP burns, and other metrics.

Mock Output Example

Bundle Summary:
• Bundle 1:
• % Supply: 15%
• SOL Spent: 500
• Unique Wallets: 50
• Still Holding: 70% of allocation.
• Bundle 2:
• % Supply: 8%
• SOL Spent: 250
• Unique Wallets: 25
• Still Holding: 50% of allocation.

Sniper Summary:
• Sniper Wallets:
• Wallet A: Acquired 2% supply in <5 seconds, dumped 80% within 2 hours.
• Wallet B: Bought directly from a secondary market at launch, now holding 1.5%.

Bubble Map Insights:
• Cluster 1: 20 wallets controlling 30% of supply.
• Cluster 2: Single wallet holding 8%, suspected of sniper behavior.

Additional Checks:
• Mint Authority: Verified.
• Freeze Authority: Verified.
• LP Burn: No suspicious activity detected.

Next Steps
• Integrate with Bitquery APIs to pull pump.fun data.
• Define thresholds for sniper detection (e.g., transaction speed, % supply).
• Build a dashboard with dynamic visualizations for bundle and sniper analysis.
• Iterate based on feedback and expand with “due diligence” checks (e.g., tokenomics validation, LP analysis).

[slimeonmyhead]

@Tejoyasaswi-17 a sniper would be someone who bought within the first few blocks after the LP was created. This usually is inclusive with bundles but could also be a single wallet/bot that bought tokens quickly.
I believe PF identifies most of this with a warning icon on their trades view

[QuarkGrandNagus]

import asyncio
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Confirmed
import pandas as pd
import numpy as np
from typing import List, Dict
import heapq

class SolanaTokenSniperAnalyzer:
def init(self, mint_address: str, solana_rpc_url: str):
self.mint_address = mint_address
self.solana_client = AsyncClient(solana_rpc_url)
self.token_holders = {}
self.bundles = []

async def fetch_token_holders(self):
    """
    Fetch all token holders for the given mint address
    """
    try:
        # Fetch token accounts for the mint
        token_accounts = await self.solana_client.get_token_largest_accounts(
            self.mint_address, 
            commitment=Confirmed
        )
        
        for account in token_accounts['result']['value']:
            self.token_holders[account['address']] = {
                'balance': account['amount'],
                'percentage': float(account['amount']) / account['total_supply']
            }
    except Exception as e:
        print(f"Error fetching token holders: {e}")

async def analyze_transaction_history(self):
    """
    Analyze transaction history to identify potential sniper bundles
    """
    try:
        # Fetch transaction signatures related to the token
        signatures = await self.solana_client.get_signatures_for_address(
            self.mint_address, 
            limit=1000
        )

        bundles = self.identify_sniper_bundles(signatures)
        return bundles
    except Exception as e:
        print(f"Error analyzing transaction history: {e}")

def identify_sniper_bundles(self, signatures):
    """
    Identify potential sniper bundles based on transaction patterns
    """
    bundles = []
    
    # Grouping logic for snipers
    wallet_timestamps = {}
    for sig in signatures:
        wallet = sig['pubkey']
        timestamp = sig['blockTime']
        
        if wallet not in wallet_timestamps:
            wallet_timestamps[wallet] = []
        
        wallet_timestamps[wallet].append(timestamp)
    
    # Detect bundles based on close timestamp clustering
    for wallet, timestamps in wallet_timestamps.items():
        if len(timestamps) > 1:
            # Check timestamp proximity
            timestamps.sort()
            bundle_candidates = self.cluster_timestamps(timestamps)
            
            for bundle in bundle_candidates:
                bundle_info = self.analyze_bundle(wallet, bundle)
                bundles.append(bundle_info)
    
    return bundles

def cluster_timestamps(self, timestamps, max_time_diff=60):
    """
    Cluster timestamps into potential sniper bundles
    """
    bundles = []
    current_bundle = [timestamps[0]]
    
    for ts in timestamps[1:]:
        if ts - current_bundle[-1] <= max_time_diff:
            current_bundle.append(ts)
        else:
            if len(current_bundle) > 1:
                bundles.append(current_bundle)
            current_bundle = [ts]
    
    if len(current_bundle) > 1:
        bundles.append(current_bundle)
    
    return bundles

def analyze_bundle(self, primary_wallet, timestamps):
    """
    Analyze details of a potential sniper bundle
    """
    bundle_info = {
        'primary_wallet': primary_wallet,
        'timestamps': timestamps,
        'unique_wallets': set(),
        'total_sol_spent': 0,
        'total_tokens_held': 0,
        'percentage_of_supply': 0
    }
    
    # Additional bundle analysis logic here
    
    return bundle_info

def generate_bundle_report(self, bundles):
    """
    Generate a comprehensive report of identified sniper bundles
    """
    bundle_data = []
    for bundle in bundles:
        bundle_data.append({
            'Primary Wallet': bundle['primary_wallet'],
            'Timestamps': bundle['timestamps'],
            'Unique Wallets': len(bundle['unique_wallets']),
            'Total SOL Spent': bundle['total_sol_spent'],
            'Total Tokens Held': bundle['total_tokens_held'],
            'Percentage of Supply': bundle['percentage_of_supply']
        })
    
    df = pd.DataFrame(bundle_data)
    return df

Example usage

async def main():
analyzer = SolanaTokenSniperAnalyzer(
mint_address='YourMintAddressHere',
solana_rpc_url='https://api.mainnet-beta.solana.com'
)
await analyzer.fetch_token_holders()
bundles = await analyzer.analyze_transaction_history()
report = analyzer.generate_bundle_report(bundles)
print(report)

Run the example

asyncio.run(main())

[slimeonmyhead]

@QuarkGrandNagus this looks pretty good, can you do this in Typescript and make it an agent tool?

[QuarkGrandNagus]

sure thing

[Tejoyasaswi-17]

Should we create a new navigation. I have created a new navigation under memories called tools. It has this new tool for flagging snipers given the solana mint address. In this tools itself we can extend further(Complete due diligence report, OG request features) and add more tools so that it kind of becomes end-end analysis suite.

[QuarkGrandNagus]

@QuarkGrandNagus this looks pretty good, can you do this in Typescript and make it an agent tool?

interface TokenHolder {
balance: string;
percentage: number;
}

interface Bundle {
primary_wallet: string;
timestamps: number[];
unique_wallets: Set;
total_sol_spent: number;
total_tokens_held: number;
percentage_of_supply: number;
}

interface BundleReport {
'Primary Wallet': string;
'Timestamps': number[];
'Unique Wallets': number;
'Total SOL Spent': number;
'Total Tokens Held': number;
'Percentage of Supply': number;
}

interface SignatureInfo {
pubkey: string;
blockTime: number;
}

class SniperDetector {
private mint_address: string;
private solana_client: any; // Replace with proper Solana client type
private token_holders: Record<string, TokenHolder>;

constructor(mint_address: string, solana_client: any) {
    this.mint_address = mint_address;
    this.solana_client = solana_client;
    this.token_holders = {};
}

async fetchTokenHolders(): Promise<void> {
    try {
        // Fetch token accounts for the mint
        const token_accounts = await this.solana_client.getTokenLargestAccounts(
            this.mint_address,
            { commitment: 'confirmed' }
        );

        for (const account of token_accounts.result.value) {
            this.token_holders[account.address] = {
                balance: account.amount,
                percentage: parseFloat(account.amount) / account.total_supply
            };
        }
    } catch (e) {
        console.error(`Error fetching token holders: ${e}`);
    }
}

async analyzeTransactionHistory(): Promise<Bundle[]> {
    try {
        // Fetch transaction signatures related to the token
        const signatures = await this.solana_client.getSignaturesForAddress(
            this.mint_address,
            { limit: 1000 }
        );
        const bundles = this.identifySniperBundles(signatures);
        return bundles;
    } catch (e) {
        console.error(`Error analyzing transaction history: ${e}`);
        return [];
    }
}

private identifySniperBundles(signatures: SignatureInfo[]): Bundle[] {
    const bundles: Bundle[] = [];
    
    // Grouping logic for snipers
    const wallet_timestamps: Record<string, number[]> = {};
    for (const sig of signatures) {
        const wallet = sig.pubkey;
        const timestamp = sig.blockTime;
        
        if (!wallet_timestamps[wallet]) {
            wallet_timestamps[wallet] = [];
        }
        
        wallet_timestamps[wallet].push(timestamp);
    }
    
    // Detect bundles based on close timestamp clustering
    for (const [wallet, timestamps] of Object.entries(wallet_timestamps)) {
        if (timestamps.length > 1) {
            // Check timestamp proximity
            timestamps.sort((a, b) => a - b);
            const bundle_candidates = this.clusterTimestamps(timestamps);
            
            for (const bundle of bundle_candidates) {
                const bundle_info = this.analyzeBundle(wallet, bundle);
                bundles.push(bundle_info);
            }
        }
    }
    
    return bundles;
}

private clusterTimestamps(timestamps: number[], max_time_diff: number = 60): number[][] {
    const bundles: number[][] = [];
    let current_bundle: number[] = [timestamps[0]];
    
    for (const ts of timestamps.slice(1)) {
        if (ts - current_bundle[current_bundle.length - 1] <= max_time_diff) {
            current_bundle.push(ts);
        } else {
            if (current_bundle.length > 1) {
                bundles.push([...current_bundle]);
            }
            current_bundle = [ts];
        }
    }
    
    if (current_bundle.length > 1) {
        bundles.push(current_bundle);
    }
    
    return bundles;
}

private analyzeBundle(primary_wallet: string, timestamps: number[]): Bundle {
    const bundle_info: Bundle = {
        primary_wallet,
        timestamps,
        unique_wallets: new Set<string>(),
        total_sol_spent: 0,
        total_tokens_held: 0,
        percentage_of_supply: 0
    };
    
    // Additional bundle analysis logic here
    
    return bundle_info;
}

generateBundleReport(bundles: Bundle[]): BundleReport[] {
    return bundles.map(bundle => ({
        'Primary Wallet': bundle.primary_wallet,
        'Timestamps': bundle.timestamps,
        'Unique Wallets': bundle.unique_wallets.size,
        'Total SOL Spent': bundle.total_sol_spent,
        'Total Tokens Held': bundle.total_tokens_held,
        'Percentage of Supply': bundle.percentage_of_supply
    }));
}

}

[QuarkGrandNagus]

@QuarkGrandNagus this looks pretty good, can you do this in Typescript and make it an agent tool?

committed these changes

import { Connection, PublicKey } from '@solana/web3.js';
import { Agent } from 'agent';

interface TokenHolder {
balance: string;
percentage: number;
}

interface Bundle {
primary_wallet: string;
timestamps: number[];
unique_wallets: Set;
total_sol_spent: number;
total_tokens_held: number;
percentage_of_supply: number;
}

interface BundleReport {
'Primary Wallet': string;
'Timestamps': number[];
'Unique Wallets': number;
'Total SOL Spent': number;
'Total Tokens Held': number;
'Percentage of Supply': number;
}

interface SignatureInfo {
pubkey: string;
blockTime: number;
}

class SniperDetectorAgent extends Agent {
private mint_address: string;
private solana_client: Connection;

constructor(mint_address: string, solana_client: Connection) {
super();
this.mint_address = mint_address;
this.solana_client = solana_client;
}

async start() {
await this.fetchTokenHolders();
const bundles = await this.analyzeTransactionHistory();
const bundleReports = this.generateBundleReport(bundles);
console.log(bundleReports);
}

async fetchTokenHolders(): Promise {
try {
const token_accounts = await this.solana_client.getTokenLargestAccounts(
new PublicKey(this.mint_address),
{ commitment: 'confirmed' }
);

    // Populate token_holders object  
    for (const account of token_accounts.result.value) {  
       // ...  
    }  
  } catch (e) {  
    console.error(`Error fetching token holders: ${e}`);  
  }  

}

async analyzeTransactionHistory(): Promise<Bundle[]> {
try {
const signatures = await this.solana_client.getSignaturesForAddress(
new PublicKey(this.mint_address),
{ limit: 1000 }
);
const bundles = this.identifySniperBundles(signatures);
return bundles;
} catch (e) {
console.error(Error analyzing transaction history: ${e});
return [];
}
}

private identifySniperBundles(signatures: SignatureInfo[]): Bundle[] {
const bundles: Bundle[] = [];

  // Grouping logic for snipers  
  const wallet_timestamps: Record<string, number[]> = {};  
  for (const sig of signatures) {  
    const wallet = sig.pubkey;  
    const timestamp = sig.blockTime;  

    if (!wallet_timestamps[wallet]) {  
       wallet_timestamps[wallet] = [];  
    }  

    wallet_timestamps[wallet].push(timestamp);  
  }  

  // Detect bundles based on close timestamp clustering  
  for (const [wallet, timestamps] of Object.entries(wallet_timestamps)) {  
    if (timestamps.length > 1) {  
       // Check timestamp proximity  
       timestamps.sort((a, b) => a - b);  
       const bundle_candidates = this.clusterTimestamps(timestamps);  

       for (const bundle of bundle_candidates) {  
          const bundle_info = this.analyzeBundle(wallet, bundle);  
          bundles.push(bundle_info);  
       }  
    }  
  }  

  return bundles;  

}

private clusterTimestamps(timestamps: number[], max_time_diff: number = 60): number[][] {
const bundles: number[][] = [];
let current_bundle: number[] = [timestamps[0]];

  for (const ts of timestamps.slice(1)) {  
    if (ts - current_bundle[current_bundle.length - 1] <= max_time_diff) {  
       current_bundle.push(ts);  
    } else {  
       if (current_bundle.length > 1) {  
          bundles.push([...current_bundle]);  
       }  
       current_bundle = [ts];  
    }  
  }  

  if (current_bundle.length > 1) {  
    bundles.push(current_bundle);  
  }  

  return bundles;  

}

private analyzeBundle(primary_wallet: string, timestamps: number[]): Bundle {
const bundle_info: Bundle = {
primary_wallet,
timestamps,
unique_wallets: new Set(),
total_sol_spent: 0,
total_tokens_held: 0,
percentage_of_supply: 0
};

  // Additional bundle analysis logic here  
  // ...  

  return bundle_info;  

}

generateBundleReport(bundles: Bundle[]): BundleReport[] {
if (bundles.length === 0) {
return [];
}

  return bundles.map(bundle => ({  
    'Primary Wallet': bundle.primary_wallet,  
    'Timestamps': bundle.timestamps,  
    'Unique Wallets': bundle.unique_wallets.size,  
    'Total SOL Spent': bundle.total_sol_spent,  
    'Total Tokens Held': bundle.total_tokens_held,  
    'Percentage of Supply': bundle.percentage_of_supply  
  }));  

}
}

// Create a new instance of the agent
const mint_address = 'your_mint_address';
const solana_client = new Connection('https://api.devnet.solana.com');
const agent = new SniperDetectorAgent(mint_address, solana_client);

// Start the agent
agent.start();

[slimeonmyhead]

@Tejoyasaswi-17
The tools are injected into the agent, see the example here of a basic tool that asks the user for confirmation: https://github.com/NeurProjects/neur-app/blob/main/src/ai/generic/util.tsx

[Ireyemi123]

Tools are injected into the agent