Skip to main content

Monitor Solana Liquidity Pools with Yellowstone gRPC Geyser Plugin (Go)

Updated on
Apr 25, 2025
All
Solana
Go
Geyser
Marketplace

9 min read

Overview​

Yellowstone Dragon's Mouth (Yellowstone) is a powerful gRPC interface built on Solana's Geyser plugin system that streams blockchain data in real-time. By connecting to this service, you can track activity across multiple DEX pools simultaneously, analyze transaction patterns, and build responsive applications that require up-to-the-millisecond data.

In this guide, you will:

  1. Set up a Go environment for Yellowstone gRPC
  2. Create a client to subscribe to SOL/USDC pool transactions
  3. Monitor transactions across an active Raydium liquidity pool
  4. Analyze transaction throughput with basic statistics

Prerequisites​


Understanding Yellowstone and Geyser​

What is Geyser?​

Geyser is Solana's plugin system that enables validators to stream real-time blockchain data to external systems without imposing heavy RPC load. Instead of repeatedly polling the blockchain through RPC calls, Geyser pushes data as it becomes available, significantly reducing latency and resource usage.

What is Yellowstone?​

Yellowstone Dragon's Mouth is an open-source gRPC interface built on Solana's Geyser plugin. It delivers high-performance, type-safe streaming of:


  • Account updates
  • Transactions
  • Entries
  • Block notifications
  • Slot notifications

For DeFi applications and trading systems, this real-time data access can provide a critical competitive edge.

Setting Up Your Environment​

First, let's create a new Go project and install the necessary dependencies.


  1. Create a project directory:
mkdir solana-dex-monitor && cd solana-dex-monitor

  1. Initialize the Go module:
go mod init solana-dex-monitor

  1. Install required dependencies:
go get google.golang.org/grpc
go get github.com/joho/godotenv
go get github.com/mr-tron/base58
go get github.com/rpcpool/yellowstone-grpc/examples/golang/proto

  1. Create a .env file to store your QuickNode credentials:
qn_grpc_url=your-quicknode-yellowstone-endpoint.grpc.solana.quiknode.pro:443
qn_grpc_token=your-quicknode-token

You can find information on configuring your endpoint in our docs, here.

Building the Liquidity Pool Monitor​

Now, let's create our main application to monitor liquidity pool transactions. For this example, we will utilize a very active SOL/USDC pool on Raydium, 3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv, but you can easily add additional pools or adapt this to other pools or DEXes.

Create a file named main.go and add the following code:

package main

import (
	"context"
	"crypto/tls"
	"fmt"
	"log"
	"os"
	"strings"
	"sync"
	"time"

	pb "github.com/rpcpool/yellowstone-grpc/examples/golang/proto"

	"github.com/joho/godotenv"
	"github.com/mr-tron/base58"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/encoding/gzip"
	"google.golang.org/grpc/keepalive"
)

var (
	endpoint string
	token    string
)

var SolUsdcPoolAddresses = []string{
	"3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv", // Example Raydium SOL/USDC pool
	// Add more pool addresses as needed
}

// Load environment variables
func init() {
	err := godotenv.Load()
	if err != nil {
		log.Fatalf("Error loading .env file: %v", err)
	}

	endpoint = getEnv("qn_grpc_url", "example.com:10000") // Default value as fallback
	token = getEnv("qn_grpc_token", "token")
}

// Helper function to get environment variable with a default value
func getEnv(key, defaultValue string) string {
	value := os.Getenv(key)
	if value == "" {
		return defaultValue
	}
	return value
}

// PoolTxStats tracks statistics for transactions
type PoolTxStats struct {
	txCount     int
	firstTxTime time.Time
	lastTxTime  time.Time
	mutex       sync.Mutex
}

// Global variables for tracking statistics
var (
	poolStats  = make(map[string]*PoolTxStats) // Key is slot as string
	statsMutex sync.RWMutex
)

type tokenAuth struct {
	token string
}

func (t tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
		"authorization": t.token,
	}, nil
}

func (t tokenAuth) RequireTransportSecurity() bool {
	return true
}

// Function to safely extract signature from transaction
func extractSignature(tx *pb.SubscribeUpdateTransaction) string {
	if tx == nil {
		return "No transaction"
	}

	if sig := tx.GetTransaction().GetSignature(); len(sig) > 0 {
		// Convert the binary signature to base58
		return base58.Encode(sig)
	}

	return "No signature found"
}

func boolPtr(b bool) *bool {
	return &b
}

func main() {
	// Setup connection parameters
	kacp := keepalive.ClientParameters{
		Time:                10 * time.Second,
		Timeout:             5 * time.Second,
		PermitWithoutStream: true,
	}

	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})),
		grpc.WithKeepaliveParams(kacp),
		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(1024*1024*1024), grpc.UseCompressor(gzip.Name)),
		grpc.WithPerRPCCredentials(tokenAuth{token: token}),
	}

	// Establish connection
	conn, err := grpc.Dial(endpoint, opts...)
	if err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer conn.Close()

	client := pb.NewGeyserClient(conn)

	// Create subscription request for SOL/USDC liquidity pool transactions
	transactions := make(map[string]*pb.SubscribeRequestFilterTransactions)
	transactions["sol_usdc_pool_txs"] = &pb.SubscribeRequestFilterTransactions{
		Vote:            boolPtr(false),
		Failed:          boolPtr(false),
		AccountInclude:  SolUsdcPoolAddresses,
		AccountExclude:  []string{},
		AccountRequired: []string{}, 
	}

	commitment := pb.CommitmentLevel_CONFIRMED
	subReq := &pb.SubscribeRequest{
		Transactions: transactions,
		Commitment:   &commitment,
	}

	fmt.Println("Connecting to Yellowstone gRPC...")
	stream, err := client.Subscribe(context.Background())
	if err != nil {
		fmt.Printf("Failed to subscribe to yellowstone: %v\n", err)
		return
	}

	fmt.Println("Sending subscription request for SOL/USDC pool transactions...")
	if err = stream.Send(subReq); err != nil {
		fmt.Printf("Failed to send subscription request: %v\n", err)
		return
	}

	// Print header for pool statistics
	fmt.Printf("\n%-12s %-12s %-15s %-20s\n",
		"Slot", "TX Count", "TX/sec", "Total Time (ms)")
	fmt.Println(strings.Repeat("-", 65))

	// Print stats every 5 seconds
	lastPrintTime := time.Now()
	printInterval := time.Second * 5

	fmt.Println("Monitoring SOL/USDC liquidity pool transactions...")
	for {
		m, err := stream.Recv()
		if err != nil {
			fmt.Printf("Failed to receive yellowstone message: %v\n", err)
			return
		}

		if tx := m.GetTransaction(); tx != nil {
			// Extract the transaction signature
			signature := extractSignature(tx)

			// Process the transaction using the slot to keep track
			slot := tx.GetSlot()
			now := time.Now()

			// Maintain a simplified tracking approach based on slot
			statsMutex.Lock()
			slotStr := fmt.Sprintf("%d", slot)
			if _, exists := poolStats[slotStr]; !exists {
				poolStats[slotStr] = &PoolTxStats{
					firstTxTime: now,
					lastTxTime:  now,
				}
			}

			stats := poolStats[slotStr]
			stats.mutex.Lock()
			stats.txCount++
			stats.lastTxTime = now
			stats.mutex.Unlock()
			statsMutex.Unlock()

			// Print transaction info with signature
			fmt.Printf("Pool transaction detected at Slot=%d\nSignature=%s\n", slot, signature)

			// Print statistics periodically
			if now.Sub(lastPrintTime) >= printInterval {
				printPoolStats()
				lastPrintTime = now

				// Cleanup old statistics
				cleanupOldStats()
			}
		}
	}
}

func printPoolStats() {
	statsMutex.RLock()
	defer statsMutex.RUnlock()

	fmt.Println("\nTransaction Statistics:")
	fmt.Printf("\n%-12s %-12s %-15s %-20s\n",
		"Slot", "TX Count", "TX/sec", "Total Time (ms)")
	fmt.Println(strings.Repeat("-", 65))

	for slotStr, stats := range poolStats {
		stats.mutex.Lock()
		duration := stats.lastTxTime.Sub(stats.firstTxTime).Milliseconds()
		var txPerSec float64
		if duration > 0 {
			txPerSec = float64(stats.txCount) / (float64(duration) / 1000.0)
		}

		fmt.Printf("%-12s %-12d %-15.2f %-20d\n",
			slotStr,
			stats.txCount,
			txPerSec,
			duration,
		)
		stats.mutex.Unlock()
	}
	fmt.Println()
}

func cleanupOldStats() {
	statsMutex.Lock()
	defer statsMutex.Unlock()

	now := time.Now()
	for slotStr, stats := range poolStats {
		stats.mutex.Lock()
		// Remove statistics older than 1 minute
		if now.Sub(stats.lastTxTime) > time.Minute {
			delete(poolStats, slotStr)
		}
		stats.mutex.Unlock()
	}
}

Understanding the Code​

Let's break down the key components of our application:

Configuration and Setup​

  1. Pool Addresses: We define a list of example SOL/USDC pool addresses--for this example, we are using an active Raydium pool. You can add more pools by appending to the SolUsdcPoolAddresses slice.

  2. Authentication: The tokenAuth struct handles authorization with your QuickNode token.

  3. Connection Setup: We configure gRPC connection settings with TLS, compression, and keepalive parameters.

Transaction Subscription​

The core of our monitoring system is the subscription request:

transactions["sol_usdc_pool_txs"] = &pb.SubscribeRequestFilterTransactions{
    Vote:            boolPtr(false),
    Failed:          boolPtr(false),
    AccountInclude:  SolUsdcPoolAddresses,
    AccountExclude:  []string{},
    AccountRequired: []string{}, 
}

This filter:

  • Excludes vote transactions
  • Only includes successful transactions
  • Monitors activity involving either the DEX program IDs or specific pool accounts

Transaction Processing​

For each incoming transaction:

  1. We extract the signature using the extractSignature function
  2. Track transaction counts and timing by slot
  3. Periodically calculate and display statistics (transactions per second, total duration)
  4. Clean up old statistics to manage memory usage

Running the Monitor​

To run the monitor:

go run main.go

You should see output similar to:

Connecting to Yellowstone gRPC...
Sending subscription request for SOL/USDC pool transactions...

Slot        TX Count     TX/sec         Total Time (ms)     
-----------------------------------------------------------------
Monitoring SOL/USDC liquidity pool transactions...
Pool transaction detected at Slot=212439883
Signature=4ZV7JsQTwQfLWtN9YMu2EJkTKjyAC9Yjjd1TGY8X5qvqYhfpfTKk3PUK5NZ2P9HFfxXUE2mRJsW2LcUfTF1oTBcP

Transaction Statistics:

Slot        TX Count     TX/sec         Total Time (ms)     
-----------------------------------------------------------------
212439883   1            0.20           5000

Enhancing the Monitor​

Want to keep building? Here are some ideas to enhance your monitor:

1. Add More Pools​

You can easily add more pools to monitor by including their addresses in the SolUsdcPoolAddresses slice:

var SolUsdcPoolAddresses = []string{
	"3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv", // Example Raydium SOL/USDC pool
	// Add more pool addresses as needed
}

2. Parse Transaction Data​

Instead of just counting transactions, you could decode transaction data to extract swap amounts, price impacts, and other details:

// Add this function to decode transaction data
func decodeTransaction(tx *pb.SubscribeUpdateTransaction) {
    // Parse transaction data based on the program ID
    // Different DEXes (Orca, Raydium, Jupiter) have different transaction structures
}

3. Store Data for Analysis​

Connect the monitor to a database to store transaction data for later analysis:

// Add database integration
func storeTransactionData(slot uint64, signature string, details map[string]interface{}) {
    // Insert into database (PostgreSQL, InfluxDB, etc.)
}

4. Implement Price Alerts​

After parsing the account data, you could add logic to detect significant price movements or unusual activity:

// Add price monitoring
func detectPriceAnomaly(currentPrice, previousPrice float64) bool {
    // Implement anomaly detection logic
    percentChange := (currentPrice - previousPrice) / previousPrice * 100
    return math.Abs(percentChange) > 1.0 // Alert on 1% price change
}

Wrap Up​

Using Yellowstone gRPC with Go provides a powerful way to monitor Solana liquidity pools with extremely low latency. This approach enables real-time tracking of DEX activity across multiple programs simultaneously, giving you the data you need for trading, analytics, or monitoring applications.

For trading bots or arbitrage systems where milliseconds matter, Yellowstone's streaming approach provides a significant advantage over traditional RPC methods. By directly tapping into the Solana blockchain data firehose, you can stay informed of market movements as they happen.

We ❤️ Feedback!

Let us know if you have any feedback or requests for new topics. We'd love to hear from you.

Additional Resources​

If you have any questions or need help implementing your Solana dApp, join our Discord community or reach out to our support team!

Written by

Aaron Milano

Last updated on
Share this guide