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は、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=your-quicknode-token

エンドポイントの設定に関する情報は、こちらのドキュメントでご確認いただけます。

流動性プールモニターの構築

それでは、流動性プールの取引を監視するメインアプリケーションを作成しましょう。この例では、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 struct は、Quicknode トークンを使用して認証処理を行います。
接続の設定：TLS、圧縮、キープアライブのパラメータを使用して、gRPCの接続設定を行います。

トランザクションの購読

当社の監視システムの中核となるのは、サブスクリプションリクエストです：

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

このフィルター：

投票取引は除く
正常に完了した取引のみを含みます
DEXプログラムIDまたは特定のプールアカウントのいずれかに関連するアクティビティを監視します

トランザクション処理

受信する各トランザクションについて：

我々は、以下の方法を用いてシグネチャを抽出する。 extractSignature 関数
スロットごとの取引件数とタイミングを追跡する
統計情報（1秒あたりのトランザクション数、合計所要時間）を定期的に計算して表示する
メモリ使用量を管理するために、古い統計情報を削除する

モニターの実行

モニターを実行するには：

main.go を実行してください

次のような出力が表示されるはずです：

Solana gRPC に接続中...
SOL/USDCプールのトランザクションに対するサブスクリプションリクエストを送信中...

スロット        トランザクション数     トランザクション/秒         合計時間 (ms)     
-----------------------------------------------------------------
SOL/USDC 流動性プールのトランザクションを監視中...
スロット=212439883でプールトランザクションを検出
署名=4ZV7JsQTwQfLWtN9YMu2EJkTKjyAC9Yjjd1TGY8X5qvqYhfpfTKk3PUK5NZ2P9HFfxXUE2mRJsW2LcUfTF1oTBcP

トランザクション統計:

スロット        トランザクション数     トランザクション/秒         合計時間 (ms)     
-----------------------------------------------------------------
212439883   1            0.20           5000

モニターの機能強化

さらにカスタマイズしたいですか？モニターをさらに使いやすくするためのアイデアをいくつかご紹介します：

1. プールをさらに増やす

監視対象のプールを追加するには、そのアドレスを SolUsdcPoolAddresses slice:

var SolUsdcPoolAddresses = []文字列{
	"3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv", // RaydiumのSOL/USDCプールの例
	// 必要に応じてプールアドレスを追加してください
}

2. 取引データの解析

単に取引件数を数えるだけでなく、取引データを解析して、スワップ額や価格への影響、その他の詳細情報を抽出することも可能です：

// トランザクションデータをデコードするための関数を追加
func decodeTransaction(tx *pb.SubscribeUpdateTransaction) {
    // プログラムIDに基づいてトランザクションデータを解析する
    // DEX（Orca、Raydium、Jupiter）ごとにトランザクションの構造が異なる
}

3. 分析用のデータの保存

後での分析に備えて、トランザクションデータを保存するために、モニターをデータベースに接続します：

// データベース連携を追加
func storeTransactionData(slot uint64, signature string, 詳細 マップ[文字列]インターフェース{}) {
    // データベース（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について​

「ガイザー」とは何ですか？​

Solana gRPCとは何ですか？​

環境のセットアップ​

流動性プールモニターの構築​

コードの理解​

設定とセットアップ​

トランザクションの購読​

トランザクション処理​

モニターの実行​

モニターの機能強化​

1. プールをさらに増やす​

2. 取引データの解析​

3. 分析用のデータの保存​

4. 価格アラートを実装する​

まとめ​