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 이상)가 설치되어 있어야 합니다.

솔라나 gRPC와 Geyser 이해하기

‘Geyser’란 무엇인가요?

Geyser는 검증자가 과도한 RPC 부하를 유발하지 않고도 실시간 블록체인 데이터를 외부 시스템으로 스트리밍할 수 있게 해주는 솔라나의 플러그인 시스템입니다. Geyser는 RPC 호출을 통해 블록체인을 반복적으로 폴링하는 대신, 데이터가 생성되는 즉시 이를 푸시하므로 지연 시간과 리소스 사용량을 크게 줄여줍니다.

솔라나 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, 압축 및 keepalive 매개변수를 사용하여 gRPC 연결 설정을 구성합니다.

거래 구독

당사 모니터링 시스템의 핵심은 구독 요청입니다:

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

이 필터:

투표 관련 거래는 제외됩니다
성공한 거래만 포함됩니다
DEX 프로그램 ID 또는 특정 풀 계정과 관련된 활동을 모니터링합니다.

거래 처리

수신되는 각 거래에 대해:

우리는 다음을 사용하여 시그니처를 추출합니다. 서명 추출 함수
슬롯별 거래 건수 및 발생 시점 추적
통계 정보(초당 트랜잭션 수, 총 소요 시간)를 주기적으로 계산하여 표시합니다.
메모리 사용량을 관리하기 위해 오래된 통계 데이터를 정리합니다

모니터 실행하기

모니터를 실행하려면:

main.go를 실행하세요

다음과 비슷한 출력이 표시될 것입니다:

Solana gRPC에 연결 중...
SOL/USDC 풀 거래에 대한 구독 요청 전송 중...

슬롯        거래 건수     초당 거래 수         총 소요 시간 (ms)     
-----------------------------------------------------------------
SOL/USDC 유동성 풀 거래 모니터링 중...
슬롯=212439883에서 풀 거래 감지됨
서명=4ZV7JsQTwQfLWtN9YMu2EJkTKjyAC9Yjjd1TGY8X5qvqYhfpfTKk3PUK5NZ2P9HFfxXUE2mRJsW2LcUfTF1oTBcP

거래 통계:

슬롯        거래 건수     초당 거래 수         총 소요 시간 (ms)     
-----------------------------------------------------------------
212439883   1            0.20           5000

모니터 성능 향상

계속 만들어 보고 싶으신가요? 모니터를 더욱 멋지게 꾸밀 수 있는 몇 가지 아이디어를 소개합니다:

1. 수영장 더 추가하기

모니터링할 풀을 더 추가하려면 해당 주소를 SolUsdcPoolAddresses slice:

var SolUsdcPoolAddresses = []string{
	"3ucNos4NbumPLZNWztqGHNFFgkHeRMBQAVemeeomsUxv", // Raydium SOL/USDC 풀 예시
	// 필요에 따라 풀 주소를 더 추가하세요
}

2. 거래 데이터 분석

단순히 거래 건수를 집계하는 대신, 거래 데이터를 분석하여 스왑 금액, 가격 변동 영향 및 기타 세부 정보를 추출할 수 있습니다:

// 트랜잭션 데이터를 디코딩하기 위해 이 함수를 추가하세요
func decodeTransaction(tx *pb.구독 업데이트 트랜잭션) {
    // 프로그램 ID를 기반으로 트랜잭션 데이터를 파싱합니다
    // DEX마다(Orca, Raydium, Jupiter) 트랜잭션 구조가 다릅니다
}

3. 분석을 위한 데이터 저장

나중에 분석할 수 있도록 트랜잭션 데이터를 저장하기 위해 모니터를 데이터베이스에 연결하십시오:

// 데이터베이스 연동 추가
func storeTransactionData(slot uint64, 서명 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 활동을 동시에 실시간으로 추적할 수 있어, 거래, 분석 또는 애플리케이션 모니터링에 필요한 데이터를 확보할 수 있습니다.

밀리초 단위의 반응 속도가 중요한 트레이딩 봇이나 차익거래 시스템의 경우, 솔라나 gRPC 스트리밍 방식은 기존의 RPC 방식에 비해 상당한 이점을 제공합니다. 솔라나 블록체인의 데이터 스트림을 직접 활용함으로써, 시장 변동이 발생하는 즉시 이를 파악할 수 있습니다.

여러분의 피드백을 ❤️ 환영합니다!

의견이나 다루었으면 하는 새로운 주제가 있다면 알려주세요. 여러분의 의견을 기다리고 있습니다.

추가 자료

질문이 있거나 Solana dApp 구현에 도움이 필요하시면, 저희 Discord 커뮤니티에 가입하시거나 지원팀에 문의해 주세요!

Solana gRPC(Go)를 사용하여 Solana 유동성 풀 모니터링하기

개요​

필수 조건​

솔라나 gRPC와 Geyser 이해하기​

‘Geyser’란 무엇인가요?​

솔라나 gRPC란 무엇인가요?​

환경 설정하기​

유동성 풀 모니터 구축​

코드 이해하기​

구성 및 설정​

거래 구독​

거래 처리​

모니터 실행하기​

모니터 성능 향상​

1. 수영장 더 추가하기​

2. 거래 데이터 분석​

3. 분석을 위한 데이터 저장​

4. 가격 알림 기능 구현​

마무리​