Solana Pump AMM Data Streaming: gRPC Guide

// Multi-stream architecture for comprehensive monitoring
const monitoringSystem = {
  accounts: {
    // Monitor Pump program state changes
    pumpProgram: "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P",
    // Bonding curve accounts for active tokens
    bondingCurves: [] // Dynamic list
  },
  transactions: {
    // All Pump program interactions
    programTransactions: true,
    // System program for SOL transfers
    systemProgram: true,
    // Token program for SPL token operations
    tokenProgram: true
  }
};

import Client, { CommitmentLevel, SubscribeRequest } from "@triton-one/yellowstone-grpc";
// Note: Use the StreamManager class from the quickstart guide

class PumpMonitoringSystem {
  private streamManager: StreamManager;
  private analytics: PumpAnalytics;

  constructor(endpoint: string, apiKey: string) {
    this.streamManager = new StreamManager(
      endpoint,
      apiKey,
      this.handleUpdate.bind(this),
      this.handleError.bind(this)
    );
    this.analytics = new PumpAnalytics();
  }

  async start(): Promise<void> {
    // Start multiple streams for comprehensive monitoring
    await Promise.all([
      this.startAccountMonitoring(),
      this.startTransactionMonitoring()
    ]);
  }

  private async startAccountMonitoring(): Promise<void> {
    const subscribeRequest: SubscribeRequest = {
      accounts: {
        pumpAccounts: {
          account: [],
          owner: ["6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"], // Pump program
          filters: [
            // TODO: Add specific filters based on actual Pump.fun account structure
          ]
        }
      },
      commitment: CommitmentLevel.CONFIRMED,
      ping: { id: 1 }
    };

    await this.streamManager.connect(subscribeRequest);
  }

  private async startTransactionMonitoring(): Promise<void> {
    const subscribeRequest: SubscribeRequest = {
      transactions: {
        pumpTransactions: {
          accountInclude: ["6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"],
          accountExclude: [],
          accountRequired: [],
          vote: false,
          failed: false
        }
      },
      commitment: CommitmentLevel.CONFIRMED,
      ping: { id: 1 }
    };

    await this.streamManager.connect(subscribeRequest);
  }

  private handleUpdate(data: any): void {
    if (data.account) {
      this.handleAccountUpdate(data.account);
    }
    
    if (data.transaction) {
      this.handleTransactionUpdate(data.transaction);
    }
  }

  private handleAccountUpdate(accountData: any): void {
    try {
      const account = accountData.account;
      
      console.log('Account update received:', {
        pubkey: account.pubkey,
        owner: account.account.owner,
        dataLength: account.account.data?.length || 0
      });
      
      // TODO: Implement account data parsing based on Pump.fun's account structure
    } catch (error) {
      console.error('Error processing account update:', error);
    }
  }

  private handleTransactionUpdate(transactionData: any): void {
    try {
      const tx = transactionData.transaction;
      
      if (tx.meta?.err) {
        return; // Skip failed transactions
      }

      // Parse transaction for Pump operations
      const pumpOperation = PumpTransactionParser.parsePumpTransaction(tx);
      
      if (pumpOperation) {
        this.analytics.processPumpOperation(pumpOperation, tx);
      }
    } catch (error) {
      console.error('Error processing transaction update:', error);
    }
  }

  private handleError(error: any): void {
    console.error('Stream error:', error);
    // Implement error recovery logic
  }

  generateDailyReport(): void {
    this.analytics.generateDailyReport();
  }

  disconnect(): void {
    // Disconnect stream manager
    if (this.streamManager) {
      this.streamManager.disconnect();
    }
  }
}

// This demonstrates the structure - implement actual parsing based on Pump.fun's program
interface PumpOperation {
  type: string;
  user: string;
  signature: string;
  timestamp: number;
}

class PumpTransactionParser {
  private static PUMP_PROGRAM_ID = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";

  static parsePumpTransaction(tx: any): PumpOperation | null {
    try {
      const message = tx.transaction?.message;
      if (!message) return null;

      // Check if transaction involves Pump program
      const hasPumpProgram = message.instructions?.some((ix: any) => {
        const programId = message.accountKeys[ix.programIdIndex];
        return programId === this.PUMP_PROGRAM_ID;
      });

      if (!hasPumpProgram) return null;

      // Return basic transaction info - implement actual parsing here
      return {
        type: 'pump_transaction', // Determine actual operation type
        user: message.accountKeys[0], // Fee payer
        signature: tx.signature,
        timestamp: Date.now()
      };
    } catch (error) {
      console.error('Error parsing Pump transaction:', error);
      return null;
    }
  }

  // TODO: Implement metadata extraction based on actual Pump.fun transaction structure
}
}

class PumpAnalytics {
  private operations: PumpOperation[] = [];

  processPumpOperation(operation: PumpOperation, tx: any): void {
    // Store the operation
    this.operations.push(operation);
    
    console.log(`\n📊 PUMP OPERATION DETECTED`);
    console.log(`  Type: ${operation.type}`);
    console.log(`  User: ${operation.user}`);
    console.log(`  Signature: ${operation.signature}`);
    console.log(`  Timestamp: ${new Date(operation.timestamp).toISOString()}`);
    
    // TODO: Implement specific operation handling based on actual Pump.fun data structure
  }

  generateDailyReport(): void {
    const oneDayAgo = Date.now() - 24 * 60 * 60 * 1000;
    const recentOperations = this.operations.filter(op => op.timestamp > oneDayAgo);

    console.log(`\n📊 DAILY PUMP REPORT`);
    console.log(`  Total Operations: ${recentOperations.length}`);
    console.log(`  Unique Users: ${new Set(recentOperations.map(op => op.user)).size}`);
    
    // Group by operation type
    const typeCount = recentOperations.reduce((acc, op) => {
      acc[op.type] = (acc[op.type] || 0) + 1;
      return acc;
    }, {} as Record<string, number>);
    
    console.log(`\n  📈 Operations by Type:`);
    Object.entries(typeCount).forEach(([type, count]) => {
      console.log(`    ${type}: ${count}`);
    });
  }
}

// Main application entry point
async function main() {
  const pumpMonitor = new PumpMonitoringSystem(
    "your-grpc-endpoint",
    "YOUR_API_KEY"
  );

  console.log('🚀 Starting Pump.fun monitoring system...');
  console.log('📊 Monitoring: Token launches, trades, and market data');
  console.log('🔔 Alerts: Large trades, price movements, new launches\n');

  // Start the monitoring system
  await pumpMonitor.start();

  // Generate reports periodically
  setInterval(() => {
    pumpMonitor.generateDailyReport();
  }, 60 * 60 * 1000); // Every hour

  // Graceful shutdown
  process.on('SIGINT', () => {
    console.log('\n🛑 Shutting down Pump monitor...');
    pumpMonitor.disconnect();
    process.exit(0);
  });

  console.log('✅ Pump.fun monitoring system is running!');
  console.log('Press Ctrl+C to stop\n');
}

main().catch(console.error);