使用 Solana gRPC（Go）监控 Solana 流动性池

阅读时间 9 分钟

概述

Quicknode 的Solana gRPC（兼容 Yellowstone 的 Geyser gRPC）是一个基于 Solana Geyser 插件系统构建的强大 gRPC 接口，可实时流式传输区块链数据。通过连接此服务，您可以同时跟踪多个 DEX 资金池的活动，分析交易模式，并构建需要毫秒级实时数据的响应式应用程序。

在本指南中，您将：

为 Solana gRPC 配置 Go 运行环境
创建一个客户端来订阅 SOL/USDC 资金池的交易
监控活跃的 Raydium 流动性池中的交易
使用基本统计信息分析事务吞吐量

先决条件

一个具有 Solana gRPC 访问权限的Quicknode 账户（Scale 和 Business 套餐中已包含，或可通过 Build 和 Accelerate 套餐中的Solana gRPC 附加组件获取）
对 Go 编程有基本了解
您的计算机上已安装 Go（1.18 及以上版本）

了解 Solana gRPC 和 Geyser

什么是Geyser？

Geyser 是 Solana 的插件系统，它使验证者能够将实时区块链数据流式传输到外部系统，同时不会给 RPC 带来过重的负载。与通过 RPC 调用反复轮询区块链不同，Geyser 会在数据可用时立即推送，从而显著降低了延迟并减少了资源消耗。

什么是 Solana gRPC？

Solana gRPC 是一个基于 Solana 的 Geyser 插件构建的高性能、类型安全的流式接口。它支持以下内容的流式传输：

账户更新
交易
条目
屏蔽通知
插槽通知

对于去中心化金融（DeFi）应用和交易系统而言，这种实时数据访问能力能够带来至关重要的竞争优势。

配置环境

首先，让我们创建一个新的 Go 项目并安装必要的依赖项。

mkdir solana-dex-monitor && cd solana-dex-monitor

初始化 Go 模块：

go mod init solana-dex-monitor

安装所需的依赖项：

前往 google.golang.org/grpc

前往 github.com/joho/godotenv

前往获取 github.com/mr-tron/base58

前往 github.com/rpcpool/yellowstone-grpc/examples/golang/proto

创建一个 .env 用于存储您的 Quicknode 凭据的文件：

qn_grpc_url=your-quicknode-endpoint.grpc.solana.quiknode.pro:443
qn_grpc_token=您的-QuickNode-令牌

您可以在我们的文档中查看有关配置端点的信息，点击此处。

构建流动性池监控系统

现在，让我们创建一个用于监控流动性池交易的主应用程序。在本示例中，我们将使用 Raydium 上一个非常活跃的 SOL/USDC 流动性池（地址：3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv），但您也可以轻松添加其他流动性池，或将其适配到其他流动性池或去中心化交易所（DEX）。

创建一个名为 main.go 并添加以下代码：

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 Solana gRPC...")
	stream, err := client.Subscribe(context.Background())
	if err != nil {
		fmt.Printf("Failed to subscribe to Solana gRPC: %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 Solana gRPC 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()
	}
}

理解代码

让我们来分析一下我们应用程序的关键组成部分：

配置与设置

池地址: 我们定义了一份 SOL/USDC 池地址示例列表——在本例中，我们使用的是一个正在运行的 Raydium 池。您可以通过在 SolUsdcPoolAddresses slice.
身份验证: 该 tokenAuth 该结构体使用您的 Quicknode 令牌来处理授权。
连接设置：我们配置了包含 TLS、压缩和 keepalive 参数的 gRPC 连接设置。

交易订阅

我们监控系统的核心是订阅请求：

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

此过滤器：

不包括投票交易
仅包含成功的交易
监控涉及DEX项目ID或特定资金池账户的活动

交易处理

对于每笔入账交易：

我们使用 extractSignature 函数
按时间段跟踪交易数量和时间
定期计算并显示统计数据（每秒交易数、总持续时间）
清理旧统计数据以管理内存使用情况

运行监视器

要运行该监视器：

运行 main.go

您应该会看到类似以下的输出：

正在连接到 Solana gRPC...
正在发送SOL/USDC资金池交易的订阅请求……

时槽        交易数     每秒交易数         总耗时 (毫秒)     
-----------------------------------------------------------------
正在监控 SOL/USDC 流动性池交易...
在 Slot=212439883 处检测到池交易
签名=4ZV7JsQTwQfLWtN9YMu2EJkTKjyAC9Yjjd1TGY8X5qvqYhfpfTKk3PUK5NZ2P9HFfxXUE2mRJsW2LcUfTF1oTBcP

交易统计：

槽号        交易数     每秒交易数         总耗时 (毫秒)     
-----------------------------------------------------------------
212439883   1            0.20           5000

增强显示器功能

还想继续组装吗？以下是一些提升显示器性能的建议：

1. 增加更多泳池

您只需将这些池的地址添加到 SolUsdcPoolAddresses slice：

var SolUsdcPoolAddresses = []字符串{
	"3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv", // Raydium SOL/USDC 池示例
	// 根据需要添加更多池地址
}

2. 解析交易数据

与其仅仅统计交易笔数，不如对交易数据进行解析，从而提取掉期金额、价格影响及其他详细信息：

// 添加此函数以解码交易数据
func decodeTransaction(tx *pb.订阅更新事务) {
    // 根据程序 ID 解析交易数据
    // 不同的去中心化交易所（Orca、Raydium、Jupiter）具有不同的交易结构
}

3. 存储数据以供分析

将监控器连接到数据库，以存储事务数据供日后分析：

// 添加数据库集成
func storeTransactionData(slot uint64, signature 字符串, 详情 映射[字符串]接口{}) {
    // 插入数据库（PostgreSQL、InfluxDB 等）
}

4. 设置价格提醒

解析完账户数据后，您可以添加逻辑来检测显著的价格波动或异常活动：

// 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
}

总结

使用 Go 语言配合 Solana gRPC，能够以极低的延迟对 Solana 流动性池进行监控。这种方法支持同时实时跟踪多个程序中的 DEX 活动，为您提供交易、分析或监控应用程序所需的数据。

对于毫秒级响应至关重要的交易机器人或套利系统而言，Solana 的 gRPC 流式处理方式相比传统 RPC 方法具有显著优势。通过直接接入 Solana 区块链的数据流，您可以实时掌握市场动态。

我们 ❤️ 您的反馈！

如果您有任何反馈或对新主题的建议，请告诉我们。我们非常期待您的来信。

其他资源

如果您有任何疑问，或者在实施 Solana dApp 时需要帮助，欢迎加入我们的Discord 社区或联系我们的支持团队！

使用 Solana gRPC（Go）监控 Solana 流动性池

概述​

先决条件​

了解 Solana gRPC 和 Geyser​

什么是Geyser？​

什么是 Solana gRPC？​

配置环境​

构建流动性池监控系统​

理解代码​

配置与设置​

交易订阅​

交易处理​

运行监视器​

增强显示器功能​

1. 增加更多泳池​

2. 解析交易数据​

3. 存储数据以供分析​

4. 设置价格提醒​

总结​