From aee8921ad5688877c16f89beb1112d4958d078a8 Mon Sep 17 00:00:00 2001 From: ysq Date: Wed, 13 Aug 2025 23:30:59 +0800 Subject: [PATCH] feat: Add account event parser and protocol type definitions - Add account event parser (account_event_parser.rs) for account-level event handling - Introduce type definitions for pumpfun, pumpswap, raydium_amm_v4, raydium_clmm, raydium_cpmm protocols - Enhance event processor to support account event processing alongside transactions - Improve subscription system with separate transaction and account filters - Optimize parser configuration using Option types for cleaner code structure - Update examples and documentation to reflect new capabilities Changes include: - 6 new types.rs files for protocol-specific data structures - Updated event processor for account, transaction, and block meta events - Refactored subscription manager with account filtering support - Enhanced metrics and batch processing logic --- Cargo.toml | 3 +- README.md | 121 +++++++--- README_CN.md | 121 +++++++--- examples/parse_tx_events.rs | 4 +- src/main.rs | 111 +++++++-- src/streaming/common/config.rs | 2 +- src/streaming/common/metrics.rs | 60 ++--- src/streaming/event_parser/common/types.rs | 32 +++ .../event_parser/core/account_event_parser.rs | 186 +++++++++++++++ src/streaming/event_parser/core/mod.rs | 1 + src/streaming/event_parser/core/traits.rs | 74 +++--- .../event_parser/protocols/bonk/events.rs | 45 ++++ .../event_parser/protocols/bonk/parser.rs | 44 ++-- .../event_parser/protocols/bonk/types.rs | 173 +++++++++++++- .../event_parser/protocols/pumpfun/events.rs | 34 +++ .../event_parser/protocols/pumpfun/mod.rs | 1 + .../event_parser/protocols/pumpfun/parser.rs | 16 +- .../event_parser/protocols/pumpfun/types.rs | 98 ++++++++ .../event_parser/protocols/pumpswap/events.rs | 33 ++- .../event_parser/protocols/pumpswap/mod.rs | 3 +- .../event_parser/protocols/pumpswap/parser.rs | 20 +- .../event_parser/protocols/pumpswap/types.rs | 95 ++++++++ .../protocols/raydium_amm_v4/events.rs | 20 +- .../protocols/raydium_amm_v4/mod.rs | 1 + .../protocols/raydium_amm_v4/parser.rs | 28 +-- .../protocols/raydium_amm_v4/types.rs | 106 +++++++++ .../protocols/raydium_clmm/events.rs | 50 +++- .../protocols/raydium_clmm/mod.rs | 1 + .../protocols/raydium_clmm/parser.rs | 36 ++- .../protocols/raydium_clmm/types.rs | 222 ++++++++++++++++++ .../protocols/raydium_cpmm/events.rs | 35 ++- .../protocols/raydium_cpmm/mod.rs | 1 + .../protocols/raydium_cpmm/parser.rs | 24 +- .../protocols/raydium_cpmm/types.rs | 111 +++++++++ src/streaming/grpc/event_processor.rs | 73 +++++- src/streaming/grpc/stream_handler.rs | 20 +- src/streaming/grpc/subscription.rs | 45 ++-- src/streaming/grpc/types.rs | 48 +++- src/streaming/yellowstone_grpc.rs | 94 +++----- src/streaming/yellowstone_sub_system.rs | 2 +- 40 files changed, 1845 insertions(+), 349 deletions(-) create mode 100644 src/streaming/event_parser/core/account_event_parser.rs create mode 100644 src/streaming/event_parser/protocols/pumpfun/types.rs create mode 100644 src/streaming/event_parser/protocols/pumpswap/types.rs create mode 100644 src/streaming/event_parser/protocols/raydium_amm_v4/types.rs create mode 100644 src/streaming/event_parser/protocols/raydium_clmm/types.rs create mode 100644 src/streaming/event_parser/protocols/raydium_cpmm/types.rs diff --git a/Cargo.toml b/Cargo.toml index 46f865f..a73090f 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "solana-streamer-sdk" -version = "0.2.4" +version = "0.3.0" edition = "2021" authors = ["William ", "sgxiang ", "wei <1415121722@qq.com>"] repository = "https://github.com/0xfnzero/solana-streamer" @@ -29,6 +29,7 @@ spl-associated-token-account = "7.0.0" borsh = { version = "1.5.3", features = ["derive"] } serde = { version = "1.0.215", features = ["derive"] } serde_json = "1.0.134" +serde-big-array = "0.5.1" futures = "0.3.31" futures-util = "0.3.31" base64 = "0.22.1" diff --git a/README.md b/README.md index cd261cf..ec72e36 100755 --- a/README.md +++ b/README.md @@ -17,15 +17,17 @@ A lightweight Rust library for real-time event streaming from Solana DEX trading - **Raydium AMM V4**: Raydium's Automated Market Maker V4 events 5. **Unified Event Interface**: Consistent event handling across all supported protocols 6. **Event Parsing System**: Automatic parsing and categorization of protocol-specific events -7. **High Performance**: Optimized for low-latency event processing -8. **Batch Processing Optimization**: Batch processing events to reduce callback overhead -9. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, etc. -10. **Memory Optimization**: Object pooling and caching mechanisms to reduce memory allocations -11. **Flexible Configuration System**: Support for custom batch sizes, backpressure strategies, channel sizes, and other parameters -12. **Preset Configurations**: Provides high-performance, low-latency, ordered processing, and other preset configurations -13. **Backpressure Handling**: Supports blocking, dropping, retrying, ordered, and other backpressure strategies -14. **Runtime Configuration Updates**: Supports dynamic configuration parameter updates at runtime -15. **Full Function Performance Monitoring**: All subscribe_events functions support performance monitoring, automatically collecting and reporting performance metrics +7. **Account State Monitoring**: Real-time monitoring of protocol account states and configuration changes +8. **Transaction & Account Event Filtering**: Separate filtering for transaction events and account state changes +9. **High Performance**: Optimized for low-latency event processing +10. **Batch Processing Optimization**: Batch processing events to reduce callback overhead +11. **Performance Monitoring**: Built-in performance metrics monitoring, including event processing speed, etc. +12. **Memory Optimization**: Object pooling and caching mechanisms to reduce memory allocations +13. **Flexible Configuration System**: Support for custom batch sizes, backpressure strategies, channel sizes, and other parameters +14. **Preset Configurations**: Provides high-performance, low-latency, ordered processing, and other preset configurations +15. **Backpressure Handling**: Supports blocking, dropping, retrying, ordered, and other backpressure strategies +16. **Runtime Configuration Updates**: Supports dynamic configuration parameter updates at runtime +17. **Full Function Performance Monitoring**: All subscribe_events functions support performance monitoring, automatically collecting and reporting performance metrics ## Installation @@ -42,14 +44,14 @@ Add the dependency to your `Cargo.toml`: ```toml # Add to your Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.2.4" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.0" } ``` ### Use crates.io ```toml # Add to your Cargo.toml -solana-streamer-sdk = "0.2.4" +solana-streamer-sdk = "0.3.0" ``` ## Usage Examples @@ -78,31 +80,38 @@ use solana_streamer_sdk::{ event_parser::{ protocols::{ bonk::{ - parser::BONK_PROGRAM_ID, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, - BonkPoolCreateEvent, BonkTradeEvent, + parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, + BonkMigrateToCpswapEvent, BonkPlatformConfigAccountEvent, BonkPoolCreateEvent, + BonkPoolStateAccountEvent, BonkTradeEvent, }, pumpfun::{ - parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunMigrateEvent, + parser::PUMPFUN_PROGRAM_ID, PumpFunBondingCurveAccountEvent, + PumpFunCreateTokenEvent, PumpFunGlobalAccountEvent, PumpFunMigrateEvent, PumpFunTradeEvent, }, pumpswap::{ parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, - PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, + PumpSwapDepositEvent, PumpSwapGlobalConfigAccountEvent, + PumpSwapPoolAccountEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, }, raydium_amm_v4::{ + parser::RAYDIUM_AMM_V4_PROGRAM_ID, RaydiumAmmV4AmmInfoAccountEvent, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event, RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent, }, raydium_clmm::{ - parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmClosePositionEvent, - RaydiumClmmCreatePoolEvent, RaydiumClmmDecreaseLiquidityV2Event, - RaydiumClmmIncreaseLiquidityV2Event, RaydiumClmmOpenPositionV2Event, - RaydiumClmmOpenPositionWithToken22NftEvent, RaydiumClmmSwapEvent, - RaydiumClmmSwapV2Event, + parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmAmmConfigAccountEvent, + RaydiumClmmClosePositionEvent, RaydiumClmmCreatePoolEvent, + RaydiumClmmDecreaseLiquidityV2Event, RaydiumClmmIncreaseLiquidityV2Event, + RaydiumClmmOpenPositionV2Event, RaydiumClmmOpenPositionWithToken22NftEvent, + RaydiumClmmPoolStateAccountEvent, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, + RaydiumClmmTickArrayStateAccountEvent, }, raydium_cpmm::{ - parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmDepositEvent, - RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent, RaydiumCpmmWithdrawEvent, + parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmAmmConfigAccountEvent, + RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, + RaydiumCpmmPoolStateAccountEvent, RaydiumCpmmSwapEvent, + RaydiumCpmmWithdrawEvent, }, BlockMetaEvent, }, @@ -110,6 +119,7 @@ use solana_streamer_sdk::{ }, grpc::ClientConfig, shred_stream::ShredClientConfig, + yellowstone_grpc::{AccountFilter, TransactionFilter}, ShredStreamGrpc, YellowstoneGrpc, }, }; @@ -151,17 +161,19 @@ async fn test_grpc() -> Result<(), Box> { Protocol::Bonk, Protocol::RaydiumCpmm, Protocol::RaydiumClmm, + Protocol::RaydiumAmmV4, ]; println!("Protocols to monitor: {:?}", protocols); // Filter accounts let account_include = vec![ - PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID - PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID - BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID - RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID - RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID ]; let account_exclude = vec![]; let account_required = vec![]; @@ -171,12 +183,21 @@ async fn test_grpc() -> Result<(), Box> { println!("Starting subscription..."); + // 监听交易数据 + let transaction_filter = TransactionFilter { + account_include: account_include.clone(), + account_exclude, + account_required, + }; + + // 监听属于owner程序的账号数据 -> 账号事件监听 + let account_filter = AccountFilter { account: vec![], owner: account_include.clone() }; + grpc.subscribe_events_immediate( protocols, None, - account_include, - account_exclude, - account_required, + transaction_filter, + account_filter, None, callback, ) @@ -314,6 +335,46 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumAmmV4WithdrawPnlEvent => |e: RaydiumAmmV4WithdrawPnlEvent| { println!("RaydiumAmmV4WithdrawPnlEvent: {e:?}"); }, + // -------------------------- account ----------------------- + BonkPoolStateAccountEvent => |e: BonkPoolStateAccountEvent| { + println!("BonkPoolStateAccountEvent: {e:?}"); + }, + BonkGlobalConfigAccountEvent => |e: BonkGlobalConfigAccountEvent| { + println!("BonkGlobalConfigAccountEvent: {e:?}"); + }, + BonkPlatformConfigAccountEvent => |e: BonkPlatformConfigAccountEvent| { + println!("BonkPlatformConfigAccountEvent: {e:?}"); + }, + PumpSwapGlobalConfigAccountEvent => |e: PumpSwapGlobalConfigAccountEvent| { + println!("PumpSwapGlobalConfigAccountEvent: {e:?}"); + }, + PumpSwapPoolAccountEvent => |e: PumpSwapPoolAccountEvent| { + println!("PumpSwapPoolAccountEvent: {e:?}"); + }, + PumpFunBondingCurveAccountEvent => |e: PumpFunBondingCurveAccountEvent| { + println!("PumpFunBondingCurveAccountEvent: {e:?}"); + }, + PumpFunGlobalAccountEvent => |e: PumpFunGlobalAccountEvent| { + println!("PumpFunGlobalAccountEvent: {e:?}"); + }, + RaydiumAmmV4AmmInfoAccountEvent => |e: RaydiumAmmV4AmmInfoAccountEvent| { + println!("RaydiumAmmV4AmmInfoAccountEvent: {e:?}"); + }, + RaydiumClmmAmmConfigAccountEvent => |e: RaydiumClmmAmmConfigAccountEvent| { + println!("RaydiumClmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumClmmPoolStateAccountEvent => |e: RaydiumClmmPoolStateAccountEvent| { + println!("RaydiumClmmPoolStateAccountEvent: {e:?}"); + }, + RaydiumClmmTickArrayStateAccountEvent => |e: RaydiumClmmTickArrayStateAccountEvent| { + println!("RaydiumClmmTickArrayStateAccountEvent: {e:?}"); + }, + RaydiumCpmmAmmConfigAccountEvent => |e: RaydiumCpmmAmmConfigAccountEvent| { + println!("RaydiumCpmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { + println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); + }, }); } } diff --git a/README_CN.md b/README_CN.md index 314062a..7fde389 100644 --- a/README_CN.md +++ b/README_CN.md @@ -17,15 +17,17 @@ - **Raydium AMM V4**: Raydium 自动做市商 V4 事件 5. **统一事件接口**: 在所有支持的协议中保持一致的事件处理 6. **事件解析系统**: 自动解析和分类协议特定事件 -7. **高性能**: 针对低延迟事件处理进行优化 -8. **批处理优化**: 批量处理事件以减少回调开销 -9. **性能监控**: 内置性能指标监控，包括事件处理速度等 -10. **内存优化**: 对象池和缓存机制减少内存分配 -11. **灵活配置系统**: 支持自定义批处理大小、背压策略、通道大小等参数 -12. **预设配置**: 提供高性能、低延迟、有序处理等预设配置 -13. **背压处理**: 支持阻塞、丢弃、重试、有序等多种背压策略 -14. **运行时配置更新**: 支持在运行时动态更新配置参数 -15. **全函数性能监控**: 所有subscribe_events函数都支持性能监控，自动收集和报告性能指标 +7. **账户状态监控**: 实时监控协议账户状态和配置变更 +8. **交易与账户事件过滤**: 分别过滤交易事件和账户状态变化 +9. **高性能**: 针对低延迟事件处理进行优化 +10. **批处理优化**: 批量处理事件以减少回调开销 +11. **性能监控**: 内置性能指标监控，包括事件处理速度等 +12. **内存优化**: 对象池和缓存机制减少内存分配 +13. **灵活配置系统**: 支持自定义批处理大小、背压策略、通道大小等参数 +14. **预设配置**: 提供高性能、低延迟、有序处理等预设配置 +15. **背压处理**: 支持阻塞、丢弃、重试、有序等多种背压策略 +16. **运行时配置更新**: 支持在运行时动态更新配置参数 +17. **全函数性能监控**: 所有subscribe_events函数都支持性能监控，自动收集和报告性能指标 ## 安装 @@ -42,14 +44,14 @@ git clone https://github.com/0xfnzero/solana-streamer ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = { path = "./solana-streamer", version = "0.2.4" } +solana-streamer-sdk = { path = "./solana-streamer", version = "0.3.0" } ``` ### 使用 crates.io ```toml # 添加到您的 Cargo.toml -solana-streamer-sdk = "0.2.4" +solana-streamer-sdk = "0.3.0" ``` ## 使用示例 @@ -78,31 +80,38 @@ use solana_streamer_sdk::{ event_parser::{ protocols::{ bonk::{ - parser::BONK_PROGRAM_ID, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, - BonkPoolCreateEvent, BonkTradeEvent, + parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, + BonkMigrateToCpswapEvent, BonkPlatformConfigAccountEvent, BonkPoolCreateEvent, + BonkPoolStateAccountEvent, BonkTradeEvent, }, pumpfun::{ - parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunMigrateEvent, + parser::PUMPFUN_PROGRAM_ID, PumpFunBondingCurveAccountEvent, + PumpFunCreateTokenEvent, PumpFunGlobalAccountEvent, PumpFunMigrateEvent, PumpFunTradeEvent, }, pumpswap::{ parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, - PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, + PumpSwapDepositEvent, PumpSwapGlobalConfigAccountEvent, + PumpSwapPoolAccountEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, }, raydium_amm_v4::{ + parser::RAYDIUM_AMM_V4_PROGRAM_ID, RaydiumAmmV4AmmInfoAccountEvent, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event, RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent, }, raydium_clmm::{ - parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmClosePositionEvent, - RaydiumClmmCreatePoolEvent, RaydiumClmmDecreaseLiquidityV2Event, - RaydiumClmmIncreaseLiquidityV2Event, RaydiumClmmOpenPositionV2Event, - RaydiumClmmOpenPositionWithToken22NftEvent, RaydiumClmmSwapEvent, - RaydiumClmmSwapV2Event, + parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmAmmConfigAccountEvent, + RaydiumClmmClosePositionEvent, RaydiumClmmCreatePoolEvent, + RaydiumClmmDecreaseLiquidityV2Event, RaydiumClmmIncreaseLiquidityV2Event, + RaydiumClmmOpenPositionV2Event, RaydiumClmmOpenPositionWithToken22NftEvent, + RaydiumClmmPoolStateAccountEvent, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, + RaydiumClmmTickArrayStateAccountEvent, }, raydium_cpmm::{ - parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmDepositEvent, - RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent, RaydiumCpmmWithdrawEvent, + parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmAmmConfigAccountEvent, + RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, + RaydiumCpmmPoolStateAccountEvent, RaydiumCpmmSwapEvent, + RaydiumCpmmWithdrawEvent, }, BlockMetaEvent, }, @@ -110,6 +119,7 @@ use solana_streamer_sdk::{ }, grpc::ClientConfig, shred_stream::ShredClientConfig, + yellowstone_grpc::{AccountFilter, TransactionFilter}, ShredStreamGrpc, YellowstoneGrpc, }, }; @@ -146,17 +156,19 @@ async fn test_grpc() -> Result<(), Box> { Protocol::Bonk, Protocol::RaydiumCpmm, Protocol::RaydiumClmm, + Protocol::RaydiumAmmV4, ]; println!("Protocols to monitor: {:?}", protocols); // 过滤账号 let account_include = vec![ - PUMPFUN_PROGRAM_ID.to_string(), // 监听 pumpfun 程序ID - PUMPSWAP_PROGRAM_ID.to_string(), // 监听 pumpswap 程序ID - BONK_PROGRAM_ID.to_string(), // 监听 bonk 程序ID - RAYDIUM_CPMM_PROGRAM_ID.to_string(), // 监听 raydium_cpmm 程序ID - RAYDIUM_CLMM_PROGRAM_ID.to_string(), // 监听 raydium_clmm 程序ID + PUMPFUN_PROGRAM_ID.to_string(), // 监听 pumpfun 程序ID + PUMPSWAP_PROGRAM_ID.to_string(), // 监听 pumpswap 程序ID + BONK_PROGRAM_ID.to_string(), // 监听 bonk 程序ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // 监听 raydium_cpmm 程序ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // 监听 raydium_clmm 程序ID + RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // 监听 raydium_amm_v4 程序ID ]; let account_exclude = vec![]; let account_required = vec![]; @@ -166,12 +178,21 @@ async fn test_grpc() -> Result<(), Box> { println!("Starting subscription..."); + // 监听交易数据 + let transaction_filter = TransactionFilter { + account_include: account_include.clone(), + account_exclude, + account_required, + }; + + // 监听属于owner程序的账号数据 -> 账号事件监听 + let account_filter = AccountFilter { account: vec![], owner: account_include.clone() }; + grpc.subscribe_events_immediate( protocols, None, - account_include, - account_exclude, - account_required, + transaction_filter, + account_filter, None, callback, ) @@ -309,6 +330,46 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumAmmV4WithdrawPnlEvent => |e: RaydiumAmmV4WithdrawPnlEvent| { println!("RaydiumAmmV4WithdrawPnlEvent: {e:?}"); }, + // -------------------------- account ----------------------- + BonkPoolStateAccountEvent => |e: BonkPoolStateAccountEvent| { + println!("BonkPoolStateAccountEvent: {e:?}"); + }, + BonkGlobalConfigAccountEvent => |e: BonkGlobalConfigAccountEvent| { + println!("BonkGlobalConfigAccountEvent: {e:?}"); + }, + BonkPlatformConfigAccountEvent => |e: BonkPlatformConfigAccountEvent| { + println!("BonkPlatformConfigAccountEvent: {e:?}"); + }, + PumpSwapGlobalConfigAccountEvent => |e: PumpSwapGlobalConfigAccountEvent| { + println!("PumpSwapGlobalConfigAccountEvent: {e:?}"); + }, + PumpSwapPoolAccountEvent => |e: PumpSwapPoolAccountEvent| { + println!("PumpSwapPoolAccountEvent: {e:?}"); + }, + PumpFunBondingCurveAccountEvent => |e: PumpFunBondingCurveAccountEvent| { + println!("PumpFunBondingCurveAccountEvent: {e:?}"); + }, + PumpFunGlobalAccountEvent => |e: PumpFunGlobalAccountEvent| { + println!("PumpFunGlobalAccountEvent: {e:?}"); + }, + RaydiumAmmV4AmmInfoAccountEvent => |e: RaydiumAmmV4AmmInfoAccountEvent| { + println!("RaydiumAmmV4AmmInfoAccountEvent: {e:?}"); + }, + RaydiumClmmAmmConfigAccountEvent => |e: RaydiumClmmAmmConfigAccountEvent| { + println!("RaydiumClmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumClmmPoolStateAccountEvent => |e: RaydiumClmmPoolStateAccountEvent| { + println!("RaydiumClmmPoolStateAccountEvent: {e:?}"); + }, + RaydiumClmmTickArrayStateAccountEvent => |e: RaydiumClmmTickArrayStateAccountEvent| { + println!("RaydiumClmmTickArrayStateAccountEvent: {e:?}"); + }, + RaydiumCpmmAmmConfigAccountEvent => |e: RaydiumCpmmAmmConfigAccountEvent| { + println!("RaydiumCpmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { + println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); + }, }); } } diff --git a/examples/parse_tx_events.rs b/examples/parse_tx_events.rs index 222d260..c023e5d 100644 --- a/examples/parse_tx_events.rs +++ b/examples/parse_tx_events.rs @@ -27,7 +27,9 @@ async fn main() -> Result<()> { for signature in valid_signatures { println!("Starting transaction parsing: {}", signature); get_single_transaction_details(signature).await?; - println!("Transaction parsing completed: {}\n\n\n", signature); + println!("Transaction parsing completed: {}\n", signature); + println!("Visit link to compare data: \nhttps://solscan.io/tx/{}\n", signature); + println!("--------------------------------"); } Ok(()) diff --git a/src/main.rs b/src/main.rs index 19821f7..ef347ab 100755 --- a/src/main.rs +++ b/src/main.rs @@ -4,31 +4,38 @@ use solana_streamer_sdk::{ event_parser::{ protocols::{ bonk::{ - parser::BONK_PROGRAM_ID, BonkMigrateToAmmEvent, BonkMigrateToCpswapEvent, - BonkPoolCreateEvent, BonkTradeEvent, + parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent, + BonkMigrateToCpswapEvent, BonkPlatformConfigAccountEvent, BonkPoolCreateEvent, + BonkPoolStateAccountEvent, BonkTradeEvent, }, pumpfun::{ - parser::PUMPFUN_PROGRAM_ID, PumpFunCreateTokenEvent, PumpFunMigrateEvent, + parser::PUMPFUN_PROGRAM_ID, PumpFunBondingCurveAccountEvent, + PumpFunCreateTokenEvent, PumpFunGlobalAccountEvent, PumpFunMigrateEvent, PumpFunTradeEvent, }, pumpswap::{ parser::PUMPSWAP_PROGRAM_ID, PumpSwapBuyEvent, PumpSwapCreatePoolEvent, - PumpSwapDepositEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, + PumpSwapDepositEvent, PumpSwapGlobalConfigAccountEvent, + PumpSwapPoolAccountEvent, PumpSwapSellEvent, PumpSwapWithdrawEvent, }, raydium_amm_v4::{ + parser::RAYDIUM_AMM_V4_PROGRAM_ID, RaydiumAmmV4AmmInfoAccountEvent, RaydiumAmmV4DepositEvent, RaydiumAmmV4Initialize2Event, RaydiumAmmV4SwapEvent, RaydiumAmmV4WithdrawEvent, RaydiumAmmV4WithdrawPnlEvent, }, raydium_clmm::{ - parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmClosePositionEvent, - RaydiumClmmCreatePoolEvent, RaydiumClmmDecreaseLiquidityV2Event, - RaydiumClmmIncreaseLiquidityV2Event, RaydiumClmmOpenPositionV2Event, - RaydiumClmmOpenPositionWithToken22NftEvent, RaydiumClmmSwapEvent, - RaydiumClmmSwapV2Event, + parser::RAYDIUM_CLMM_PROGRAM_ID, RaydiumClmmAmmConfigAccountEvent, + RaydiumClmmClosePositionEvent, RaydiumClmmCreatePoolEvent, + RaydiumClmmDecreaseLiquidityV2Event, RaydiumClmmIncreaseLiquidityV2Event, + RaydiumClmmOpenPositionV2Event, RaydiumClmmOpenPositionWithToken22NftEvent, + RaydiumClmmPoolStateAccountEvent, RaydiumClmmSwapEvent, RaydiumClmmSwapV2Event, + RaydiumClmmTickArrayStateAccountEvent, }, raydium_cpmm::{ - parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmDepositEvent, - RaydiumCpmmInitializeEvent, RaydiumCpmmSwapEvent, RaydiumCpmmWithdrawEvent, + parser::RAYDIUM_CPMM_PROGRAM_ID, RaydiumCpmmAmmConfigAccountEvent, + RaydiumCpmmDepositEvent, RaydiumCpmmInitializeEvent, + RaydiumCpmmPoolStateAccountEvent, RaydiumCpmmSwapEvent, + RaydiumCpmmWithdrawEvent, }, BlockMetaEvent, }, @@ -36,6 +43,7 @@ use solana_streamer_sdk::{ }, grpc::ClientConfig, shred_stream::ShredClientConfig, + yellowstone_grpc::{AccountFilter, TransactionFilter}, ShredStreamGrpc, YellowstoneGrpc, }, }; @@ -72,21 +80,33 @@ async fn test_grpc() -> Result<(), Box> { Protocol::Bonk, Protocol::RaydiumCpmm, Protocol::RaydiumClmm, + Protocol::RaydiumAmmV4, ]; println!("Protocols to monitor: {:?}", protocols); // Filter accounts let account_include = vec![ - PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID - PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID - BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID - RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID - RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + PUMPFUN_PROGRAM_ID.to_string(), // Listen to pumpfun program ID + PUMPSWAP_PROGRAM_ID.to_string(), // Listen to pumpswap program ID + BONK_PROGRAM_ID.to_string(), // Listen to bonk program ID + RAYDIUM_CPMM_PROGRAM_ID.to_string(), // Listen to raydium_cpmm program ID + RAYDIUM_CLMM_PROGRAM_ID.to_string(), // Listen to raydium_clmm program ID + RAYDIUM_AMM_V4_PROGRAM_ID.to_string(), // Listen to raydium_amm_v4 program ID ]; let account_exclude = vec![]; let account_required = vec![]; + // 监听交易数据 + let transaction_filter = TransactionFilter { + account_include: account_include.clone(), + account_exclude, + account_required, + }; + + // 监听属于owner程序的账号数据 -> 账号事件监听 + let account_filter = AccountFilter { account: vec![], owner: account_include.clone() }; + println!("Starting to listen for events, press Ctrl+C to stop..."); println!("Monitoring programs: {:?}", account_include); @@ -95,9 +115,8 @@ async fn test_grpc() -> Result<(), Box> { grpc.subscribe_events_immediate( protocols, None, - account_include, - account_exclude, - account_required, + transaction_filter, + account_filter, None, callback, ) @@ -136,11 +155,11 @@ fn create_event_callback() -> impl Fn(Box) { |event: Box| { println!("🎉 Event received! Type: {:?}, ID: {}", event.event_type(), event.id()); match_event!(event, { - // block meta + // -------------------------- block meta ----------------------- BlockMetaEvent => |e: BlockMetaEvent| { println!("BlockMetaEvent: {e:?}"); }, - // bonk + // -------------------------- bonk ----------------------- BonkPoolCreateEvent => |e: BonkPoolCreateEvent| { // When using grpc, you can get block_time from each event println!("block_time: {:?}, block_time_ms: {:?}", e.metadata.block_time, e.metadata.block_time_ms); @@ -155,7 +174,7 @@ fn create_event_callback() -> impl Fn(Box) { BonkMigrateToCpswapEvent => |e: BonkMigrateToCpswapEvent| { println!("BonkMigrateToCpswapEvent: {e:?}"); }, - // pumpfun + // -------------------------- pumpfun ----------------------- PumpFunTradeEvent => |e: PumpFunTradeEvent| { println!("PumpFunTradeEvent: {e:?}"); }, @@ -165,7 +184,7 @@ fn create_event_callback() -> impl Fn(Box) { PumpFunCreateTokenEvent => |e: PumpFunCreateTokenEvent| { println!("PumpFunCreateTokenEvent: {e:?}"); }, - // pumpswap + // -------------------------- pumpswap ----------------------- PumpSwapBuyEvent => |e: PumpSwapBuyEvent| { println!("Buy event: {e:?}"); }, @@ -181,7 +200,7 @@ fn create_event_callback() -> impl Fn(Box) { PumpSwapWithdrawEvent => |e: PumpSwapWithdrawEvent| { println!("Withdraw event: {e:?}"); }, - // raydium_cpmm + // -------------------------- raydium_cpmm ----------------------- RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| { println!("RaydiumCpmmSwapEvent: {e:?}"); }, @@ -194,7 +213,7 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumCpmmWithdrawEvent => |e: RaydiumCpmmWithdrawEvent| { println!("RaydiumCpmmWithdrawEvent: {e:?}"); }, - // raydium_clmm + // -------------------------- raydium_clmm ----------------------- RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| { println!("RaydiumClmmSwapEvent: {e:?}"); }, @@ -219,7 +238,7 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumClmmOpenPositionV2Event => |e: RaydiumClmmOpenPositionV2Event| { println!("RaydiumClmmOpenPositionV2Event: {e:?}"); }, - // raydium_amm_v4 + // -------------------------- raydium_amm_v4 ----------------------- RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| { println!("RaydiumAmmV4SwapEvent: {e:?}"); }, @@ -235,6 +254,46 @@ fn create_event_callback() -> impl Fn(Box) { RaydiumAmmV4WithdrawPnlEvent => |e: RaydiumAmmV4WithdrawPnlEvent| { println!("RaydiumAmmV4WithdrawPnlEvent: {e:?}"); }, + // -------------------------- account ----------------------- + BonkPoolStateAccountEvent => |e: BonkPoolStateAccountEvent| { + println!("BonkPoolStateAccountEvent: {e:?}"); + }, + BonkGlobalConfigAccountEvent => |e: BonkGlobalConfigAccountEvent| { + println!("BonkGlobalConfigAccountEvent: {e:?}"); + }, + BonkPlatformConfigAccountEvent => |e: BonkPlatformConfigAccountEvent| { + println!("BonkPlatformConfigAccountEvent: {e:?}"); + }, + PumpSwapGlobalConfigAccountEvent => |e: PumpSwapGlobalConfigAccountEvent| { + println!("PumpSwapGlobalConfigAccountEvent: {e:?}"); + }, + PumpSwapPoolAccountEvent => |e: PumpSwapPoolAccountEvent| { + println!("PumpSwapPoolAccountEvent: {e:?}"); + }, + PumpFunBondingCurveAccountEvent => |e: PumpFunBondingCurveAccountEvent| { + println!("PumpFunBondingCurveAccountEvent: {e:?}"); + }, + PumpFunGlobalAccountEvent => |e: PumpFunGlobalAccountEvent| { + println!("PumpFunGlobalAccountEvent: {e:?}"); + }, + RaydiumAmmV4AmmInfoAccountEvent => |e: RaydiumAmmV4AmmInfoAccountEvent| { + println!("RaydiumAmmV4AmmInfoAccountEvent: {e:?}"); + }, + RaydiumClmmAmmConfigAccountEvent => |e: RaydiumClmmAmmConfigAccountEvent| { + println!("RaydiumClmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumClmmPoolStateAccountEvent => |e: RaydiumClmmPoolStateAccountEvent| { + println!("RaydiumClmmPoolStateAccountEvent: {e:?}"); + }, + RaydiumClmmTickArrayStateAccountEvent => |e: RaydiumClmmTickArrayStateAccountEvent| { + println!("RaydiumClmmTickArrayStateAccountEvent: {e:?}"); + }, + RaydiumCpmmAmmConfigAccountEvent => |e: RaydiumCpmmAmmConfigAccountEvent| { + println!("RaydiumCpmmAmmConfigAccountEvent: {e:?}"); + }, + RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| { + println!("RaydiumCpmmPoolStateAccountEvent: {e:?}"); + }, }); } } diff --git a/src/streaming/common/config.rs b/src/streaming/common/config.rs index 1ae1500..c050016 100644 --- a/src/streaming/common/config.rs +++ b/src/streaming/common/config.rs @@ -108,7 +108,7 @@ impl StreamClientConfig { channel_size: 20000, strategy: BackpressureStrategy::Drop, }, - enable_metrics: true, + enable_metrics: false, } } diff --git a/src/streaming/common/metrics.rs b/src/streaming/common/metrics.rs index ecaaa9c..69540d4 100644 --- a/src/streaming/common/metrics.rs +++ b/src/streaming/common/metrics.rs @@ -1,18 +1,19 @@ use std::sync::Arc; use tokio::sync::Mutex; -use super::constants::*; use super::config::StreamClientConfig; +use super::constants::*; /// 通用性能监控指标 #[derive(Debug, Clone)] pub struct PerformanceMetrics { + pub start_time: std::time::Instant, + pub process_count: u64, pub events_processed: u64, pub events_per_second: f64, pub average_processing_time_ms: f64, pub min_processing_time_ms: f64, pub max_processing_time_ms: f64, - pub cache_hit_rate: f64, pub last_update_time: std::time::Instant, pub events_in_window: u64, pub window_start_time: std::time::Instant, @@ -28,12 +29,13 @@ impl PerformanceMetrics { pub fn new() -> Self { let now = std::time::Instant::now(); Self { + start_time: std::time::Instant::now(), + process_count: 0, events_processed: 0, events_per_second: 0.0, average_processing_time_ms: 0.0, min_processing_time_ms: 0.0, max_processing_time_ms: 0.0, - cache_hit_rate: 0.0, last_update_time: now, events_in_window: 0, window_start_time: now, @@ -51,7 +53,7 @@ pub struct MetricsManager { impl MetricsManager { /// 创建新的性能监控管理器 pub fn new( - metrics: Arc>, + metrics: Arc>, config: Arc, stream_name: String, ) -> Self { @@ -68,14 +70,13 @@ impl MetricsManager { pub async fn print_metrics(&self) { let metrics = self.get_metrics().await; println!("📊 {} Performance Metrics:", self.stream_name); + println!(" Run Time: {:?}", metrics.start_time.elapsed()); + println!(" Process Count: {}", metrics.process_count); println!(" Events Processed: {}", metrics.events_processed); println!(" Events/Second: {:.2}", metrics.events_per_second); println!(" Avg Processing Time: {:.2}ms", metrics.average_processing_time_ms); println!(" Min Processing Time: {:.2}ms", metrics.min_processing_time_ms); println!(" Max Processing Time: {:.2}ms", metrics.max_processing_time_ms); - if metrics.cache_hit_rate > 0.0 { - println!(" Cache Hit Rate: {:.2}%", metrics.cache_hit_rate * 100.0); - } println!("---"); } @@ -88,9 +89,9 @@ impl MetricsManager { let metrics_manager = self.clone(); tokio::spawn(async move { - let mut interval = tokio::time::interval( - tokio::time::Duration::from_secs(DEFAULT_METRICS_PRINT_INTERVAL_SECONDS) - ); + let mut interval = tokio::time::interval(tokio::time::Duration::from_secs( + DEFAULT_METRICS_PRINT_INTERVAL_SECONDS, + )); loop { interval.tick().await; metrics_manager.print_metrics().await; @@ -98,6 +99,15 @@ impl MetricsManager { }); } + /// 更新处理次数 + pub async fn add_process_count(&self) { + if !self.config.enable_metrics { + return; + } + let mut metrics = self.metrics.lock().await; + metrics.process_count += 1; + } + /// 更新性能指标 pub async fn update_metrics(&self, events_processed: u64, processing_time_ms: f64) { // 检查是否启用性能监控 @@ -107,26 +117,29 @@ impl MetricsManager { let mut metrics = self.metrics.lock().await; let now = std::time::Instant::now(); - + metrics.events_processed += events_processed; metrics.events_in_window += events_processed; metrics.last_update_time = now; - + // 更新最快和最慢处理时间 - if processing_time_ms < metrics.min_processing_time_ms || metrics.min_processing_time_ms == 0.0 { + if processing_time_ms < metrics.min_processing_time_ms + || metrics.min_processing_time_ms == 0.0 + { metrics.min_processing_time_ms = processing_time_ms; } if processing_time_ms > metrics.max_processing_time_ms { metrics.max_processing_time_ms = processing_time_ms; } - + // 计算平均处理时间 if metrics.events_processed > 0 { - metrics.average_processing_time_ms = - (metrics.average_processing_time_ms * (metrics.events_processed - events_processed) as f64 + processing_time_ms) + metrics.average_processing_time_ms = (metrics.average_processing_time_ms + * (metrics.events_processed - events_processed) as f64 + + processing_time_ms) / metrics.events_processed as f64; } - + // 基于时间窗口计算每秒处理事件数 let window_duration = std::time::Duration::from_secs(DEFAULT_METRICS_WINDOW_SECONDS); if now.duration_since(metrics.window_start_time) >= window_duration { @@ -137,22 +150,11 @@ impl MetricsManager { // 如果窗口内没有事件，保持之前的速率或设为0 metrics.events_per_second = 0.0; } - + // 重置窗口 metrics.events_in_window = 0; metrics.window_start_time = now; } - - } - - /// 更新缓存命中率 - pub async fn update_cache_hit_rate(&self, hit_rate: f64) { - if !self.config.enable_metrics { - return; - } - - let mut metrics = self.metrics.lock().await; - metrics.cache_hit_rate = hit_rate; } /// 记录慢处理操作 diff --git a/src/streaming/event_parser/common/types.rs b/src/streaming/event_parser/common/types.rs index c22b16f..c1cbc0c 100755 --- a/src/streaming/event_parser/common/types.rs +++ b/src/streaming/event_parser/common/types.rs @@ -159,6 +159,22 @@ pub enum EventType { RaydiumAmmV4Withdraw, RaydiumAmmV4WithdrawPnl, + // Account events + AccountRaydiumAmmV4AmmInfo, + AccountPumpSwapGlobalConfig, + AccountPumpSwapPool, + AccountBonkPoolState, + AccountBonkGlobalConfig, + AccountBonkPlatformConfig, + AccountBonkVestingRecord, + AccountPumpFunBondingCurve, + AccountPumpFunGlobal, + AccountRaydiumClmmAmmConfig, + AccountRaydiumClmmPoolState, + AccountRaydiumClmmTickArrayState, + AccountRaydiumCpmmAmmConfig, + AccountRaydiumCpmmPoolState, + // Common events BlockMeta, Unknown, @@ -184,6 +200,14 @@ impl EventType { EventType::BonkInitialize => "BonkInitialize".to_string(), EventType::BonkMigrateToAmm => "BonkMigrateToAmm".to_string(), EventType::BonkMigrateToCpswap => "BonkMigrateToCpswap".to_string(), + EventType::AccountPumpFunBondingCurve => "AccountPumpFunBondingCurve".to_string(), + EventType::AccountPumpFunGlobal => "AccountPumpFunGlobal".to_string(), + EventType::AccountPumpSwapGlobalConfig => "AccountPumpSwapGlobalConfig".to_string(), + EventType::AccountPumpSwapPool => "AccountPumpSwapPool".to_string(), + EventType::AccountBonkPoolState => "AccountBonkPoolState".to_string(), + EventType::AccountBonkGlobalConfig => "AccountBonkGlobalConfig".to_string(), + EventType::AccountBonkPlatformConfig => "AccountBonkPlatformConfig".to_string(), + EventType::AccountBonkVestingRecord => "AccountBonkVestingRecord".to_string(), EventType::RaydiumCpmmSwapBaseInput => "RaydiumCpmmSwapBaseInput".to_string(), EventType::RaydiumCpmmSwapBaseOutput => "RaydiumCpmmSwapBaseOutput".to_string(), EventType::RaydiumCpmmDeposit => "RaydiumCpmmDeposit".to_string(), @@ -209,6 +233,14 @@ impl EventType { EventType::RaydiumAmmV4Initialize2 => "RaydiumAmmV4Initialize2".to_string(), EventType::RaydiumAmmV4Withdraw => "RaydiumAmmV4Withdraw".to_string(), EventType::RaydiumAmmV4WithdrawPnl => "RaydiumAmmV4WithdrawPnl".to_string(), + EventType::AccountRaydiumAmmV4AmmInfo => "AccountRaydiumAmmV4AmmInfo".to_string(), + EventType::AccountRaydiumClmmAmmConfig => "AccountRaydiumClmmAmmConfig".to_string(), + EventType::AccountRaydiumClmmPoolState => "AccountRaydiumClmmPoolState".to_string(), + EventType::AccountRaydiumClmmTickArrayState => { + "AccountRaydiumClmmTickArrayState".to_string() + } + EventType::AccountRaydiumCpmmAmmConfig => "AccountRaydiumCpmmAmmConfig".to_string(), + EventType::AccountRaydiumCpmmPoolState => "AccountRaydiumCpmmPoolState".to_string(), EventType::BlockMeta => "BlockMeta".to_string(), EventType::Unknown => "Unknown".to_string(), } diff --git a/src/streaming/event_parser/core/account_event_parser.rs b/src/streaming/event_parser/core/account_event_parser.rs new file mode 100644 index 0000000..855a394 --- /dev/null +++ b/src/streaming/event_parser/core/account_event_parser.rs @@ -0,0 +1,186 @@ +use std::collections::HashMap; +use std::sync::OnceLock; + +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType}; +use crate::streaming::event_parser::core::traits::UnifiedEvent; +use crate::streaming::event_parser::protocols::bonk::parser::BONK_PROGRAM_ID; +use crate::streaming::event_parser::protocols::pumpfun::parser::PUMPFUN_PROGRAM_ID; +use crate::streaming::event_parser::protocols::pumpswap::parser::PUMPSWAP_PROGRAM_ID; +use crate::streaming::event_parser::protocols::raydium_amm_v4::parser::RAYDIUM_AMM_V4_PROGRAM_ID; +use crate::streaming::event_parser::protocols::raydium_clmm::parser::RAYDIUM_CLMM_PROGRAM_ID; +use crate::streaming::event_parser::protocols::raydium_cpmm::parser::RAYDIUM_CPMM_PROGRAM_ID; +use crate::streaming::event_parser::Protocol; +use crate::streaming::grpc::AccountPretty; + +/// 通用事件解析器配置 +#[derive(Debug, Clone)] +pub struct AccountEventParseConfig { + pub program_id: Pubkey, + pub protocol_type: ProtocolType, + pub event_type: EventType, + pub account_discriminator: &'static [u8], + pub account_parser: AccountEventParserFn, +} + +/// 账户事件解析器 +pub type AccountEventParserFn = + fn(account: &AccountPretty, metadata: EventMetadata) -> Option>; + +static PROTOCOL_CONFIGS_CACHE: OnceLock>> = + OnceLock::new(); + +pub struct AccountEventParser {} + +impl AccountEventParser { + pub fn configs(protocols: Vec) -> Vec { + let protocols_map = PROTOCOL_CONFIGS_CACHE.get_or_init(|| { + let mut map: HashMap> = HashMap::new(); + map.insert(Protocol::PumpSwap, vec![ + AccountEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, + event_type: EventType::AccountPumpSwapGlobalConfig, + account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::GLOBAL_CONFIG_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpswap::types::global_config_parser, + }, + AccountEventParseConfig { + program_id: PUMPSWAP_PROGRAM_ID, + protocol_type: ProtocolType::PumpSwap, + event_type: EventType::AccountPumpSwapPool, + account_discriminator: crate::streaming::event_parser::protocols::pumpswap::discriminators::POOL_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpswap::types::pool_parser, + }, + ]); + map.insert(Protocol::PumpFun, vec![ + AccountEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, + event_type: EventType::AccountPumpFunBondingCurve, + account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::BONDING_CURVE_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpfun::types::bonding_curve_parser, + }, + AccountEventParseConfig { + program_id: PUMPFUN_PROGRAM_ID, + protocol_type: ProtocolType::PumpFun, + event_type: EventType::AccountPumpFunGlobal, + account_discriminator: crate::streaming::event_parser::protocols::pumpfun::discriminators::GLOBAL_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::pumpfun::types::global_parser, + }, + ]); + map.insert(Protocol::Bonk, vec![ + AccountEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, + event_type: EventType::AccountBonkPoolState, + account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::POOL_STATE_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::bonk::types::pool_state_parser, + }, + AccountEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, + event_type: EventType::AccountBonkGlobalConfig, + account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::GLOBAL_CONFIG_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::bonk::types::global_config_parser, + }, + AccountEventParseConfig { + program_id: BONK_PROGRAM_ID, + protocol_type: ProtocolType::Bonk, + event_type: EventType::AccountBonkPlatformConfig, + account_discriminator: crate::streaming::event_parser::protocols::bonk::discriminators::PLATFORM_CONFIG_ACCOUNT, + account_parser: crate::streaming::event_parser::protocols::bonk::types::platform_config_parser, + }, + ]); + map.insert(Protocol::RaydiumCpmm, vec![ + AccountEventParseConfig { + program_id: RAYDIUM_CPMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumCpmm, + event_type: EventType::AccountRaydiumCpmmAmmConfig, + account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::AMM_CONFIG, + account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::amm_config_parser, + }, + AccountEventParseConfig { + program_id: RAYDIUM_CPMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumCpmm, + event_type: EventType::AccountRaydiumCpmmPoolState, + account_discriminator: crate::streaming::event_parser::protocols::raydium_cpmm::discriminators::POOL_STATE, + account_parser: crate::streaming::event_parser::protocols::raydium_cpmm::types::pool_state_parser, + }, + ]); + map.insert(Protocol::RaydiumClmm, vec![ + AccountEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, + event_type: EventType::AccountRaydiumClmmAmmConfig, + account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::AMM_CONFIG, + account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::amm_config_parser, + }, + AccountEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, + event_type: EventType::AccountRaydiumClmmPoolState, + account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::POOL_STATE, + account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::pool_state_parser, + }, + AccountEventParseConfig { + program_id: RAYDIUM_CLMM_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumClmm, + event_type: EventType::AccountRaydiumClmmTickArrayState, + account_discriminator: crate::streaming::event_parser::protocols::raydium_clmm::discriminators::TICK_ARRAY_STATE, + account_parser: crate::streaming::event_parser::protocols::raydium_clmm::types::tick_array_state_parser, + }, + ]); + map.insert(Protocol::RaydiumAmmV4, vec![ + AccountEventParseConfig { + program_id: RAYDIUM_AMM_V4_PROGRAM_ID, + protocol_type: ProtocolType::RaydiumAmmV4, + event_type: EventType::AccountRaydiumAmmV4AmmInfo, + account_discriminator: crate::streaming::event_parser::protocols::raydium_amm_v4::discriminators::AMM_INFO, + account_parser: crate::streaming::event_parser::protocols::raydium_amm_v4::types::amm_info_parser, + }, + ]); + map + }); + + let mut configs = vec![]; + for protocol in protocols { + configs.extend(protocols_map.get(&protocol).unwrap_or(&vec![]).clone()); + } + configs + } + + pub fn parse_account_event( + protocols: Vec, + account: AccountPretty, + program_received_time_ms: i64, + ) -> Option> { + let configs = Self::configs(protocols); + for config in configs { + if account.owner == config.program_id.to_string() + && account.data[..config.account_discriminator.len()] + == *config.account_discriminator + { + let event = (config.account_parser)( + &account, + EventMetadata { + slot: account.slot, + signature: account.signature.clone(), + protocol: config.protocol_type, + event_type: config.event_type, + program_id: config.program_id, + program_received_time_ms, + ..Default::default() + }, + ); + if let Some(mut event) = event { + event.set_program_handle_time_consuming_ms( + chrono::Utc::now().timestamp_millis() - program_received_time_ms, + ); + return Some(event); + } + } + } + None + } +} diff --git a/src/streaming/event_parser/core/mod.rs b/src/streaming/event_parser/core/mod.rs index c81525e..27d61db 100755 --- a/src/streaming/event_parser/core/mod.rs +++ b/src/streaming/event_parser/core/mod.rs @@ -1,3 +1,4 @@ pub mod common_event_parser; pub mod traits; +pub mod account_event_parser; pub use traits::{EventParser, UnifiedEvent}; diff --git a/src/streaming/event_parser/core/traits.rs b/src/streaming/event_parser/core/traits.rs index d6c537a..a6d8574 100755 --- a/src/streaming/event_parser/core/traits.rs +++ b/src/streaming/event_parser/core/traits.rs @@ -542,8 +542,8 @@ pub struct GenericEventParseConfig { pub inner_instruction_discriminator: &'static str, pub instruction_discriminator: &'static [u8], pub event_type: EventType, - pub inner_instruction_parser: InnerInstructionEventParser, - pub instruction_parser: InstructionEventParser, + pub inner_instruction_parser: Option, + pub instruction_parser: Option, } /// 内联指令事件解析器 @@ -576,7 +576,7 @@ impl GenericEventParser { instruction_configs .entry(config.instruction_discriminator.to_vec()) .or_insert_with(Vec::new) - .push(config); + .push(config.clone()); } Self { program_ids, inner_instruction_configs, instruction_configs } @@ -594,21 +594,25 @@ impl GenericEventParser { program_received_time_ms: i64, index: String, ) -> Option> { - let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); - let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; - let metadata = EventMetadata::new( - signature.to_string(), - signature.to_string(), - slot, - timestamp.seconds, - block_time_ms, - config.protocol_type.clone(), - config.event_type.clone(), - config.program_id, - index, - program_received_time_ms, - ); - (config.inner_instruction_parser)(data, metadata) + if let Some(parser) = config.inner_instruction_parser { + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; + let metadata = EventMetadata::new( + signature.to_string(), + signature.to_string(), + slot, + timestamp.seconds, + block_time_ms, + config.protocol_type.clone(), + config.event_type.clone(), + config.program_id, + index, + program_received_time_ms, + ); + parser(data, metadata) + } else { + None + } } /// 通用的指令解析方法 @@ -624,21 +628,25 @@ impl GenericEventParser { program_received_time_ms: i64, index: String, ) -> Option> { - let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); - let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; - let metadata = EventMetadata::new( - signature.to_string(), - signature.to_string(), - slot, - timestamp.seconds, - block_time_ms, - config.protocol_type.clone(), - config.event_type.clone(), - config.program_id, - index, - program_received_time_ms, - ); - (config.instruction_parser)(data, account_pubkeys, metadata) + if let Some(parser) = config.instruction_parser { + let timestamp = block_time.unwrap_or(Timestamp { seconds: 0, nanos: 0 }); + let block_time_ms = timestamp.seconds * 1000 + (timestamp.nanos as i64) / 1_000_000; + let metadata = EventMetadata::new( + signature.to_string(), + signature.to_string(), + slot, + timestamp.seconds, + block_time_ms, + config.protocol_type.clone(), + config.event_type.clone(), + config.program_id, + index, + program_received_time_ms, + ); + parser(data, account_pubkeys, metadata) + } else { + None + } } } diff --git a/src/streaming/event_parser/protocols/bonk/events.rs b/src/streaming/event_parser/protocols/bonk/events.rs index 2128652..b93f2dc 100755 --- a/src/streaming/event_parser/protocols/bonk/events.rs +++ b/src/streaming/event_parser/protocols/bonk/events.rs @@ -3,6 +3,7 @@ use crate::streaming::event_parser::common::EventMetadata; use crate::streaming::event_parser::protocols::bonk::types::{ CurveParams, MintParams, PoolStatus, TradeDirection, VestingParams, }; +use crate::streaming::event_parser::protocols::bonk::{GlobalConfig, PlatformConfig, PoolState}; use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; @@ -231,6 +232,45 @@ pub struct BonkMigrateToCpswapEvent { // Macro to generate UnifiedEvent implementation, specifying the fields to be merged impl_unified_event!(BonkMigrateToCpswapEvent,); +/// 池状态 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BonkPoolStateAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub pool_state: PoolState, +} +impl_unified_event!(BonkPoolStateAccountEvent,); + +/// 全局配置 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BonkGlobalConfigAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub global_config: GlobalConfig, +} +impl_unified_event!(BonkGlobalConfigAccountEvent,); + +/// 平台配置 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BonkPlatformConfigAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub platform_config: PlatformConfig, +} +impl_unified_event!(BonkPlatformConfigAccountEvent,); + /// Event discriminator constants pub mod discriminators { // Event discriminators @@ -245,4 +285,9 @@ pub mod discriminators { pub const INITIALIZE: &[u8] = &[175, 175, 109, 31, 13, 152, 155, 237]; pub const MIGRATE_TO_AMM: &[u8] = &[207, 82, 192, 145, 254, 207, 145, 223]; pub const MIGRATE_TO_CP_SWAP: &[u8] = &[136, 92, 200, 103, 28, 218, 144, 140]; + + // 账户鉴别器 + pub const POOL_STATE_ACCOUNT: &[u8] = &[247, 237, 227, 245, 215, 195, 222, 70]; + pub const GLOBAL_CONFIG_ACCOUNT: &[u8] = &[149, 8, 156, 202, 160, 252, 176, 217]; + pub const PLATFORM_CONFIG_ACCOUNT: &[u8] = &[160, 78, 128, 0, 248, 83, 230, 160]; } diff --git a/src/streaming/event_parser/protocols/bonk/parser.rs b/src/streaming/event_parser/protocols/bonk/parser.rs index 5d34a3d..8c4d6b7 100755 --- a/src/streaming/event_parser/protocols/bonk/parser.rs +++ b/src/streaming/event_parser/protocols/bonk/parser.rs @@ -39,8 +39,8 @@ impl BonkEventParser { inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::BUY_EXACT_IN, event_type: EventType::BonkBuyExactIn, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_buy_exact_in_instruction, + inner_instruction_parser: Some(Self::parse_trade_inner_instruction), + instruction_parser: Some(Self::parse_buy_exact_in_instruction), }, GenericEventParseConfig { program_id: BONK_PROGRAM_ID, @@ -48,8 +48,8 @@ impl BonkEventParser { inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::BUY_EXACT_OUT, event_type: EventType::BonkBuyExactOut, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_buy_exact_out_instruction, + inner_instruction_parser: Some(Self::parse_trade_inner_instruction), + instruction_parser: Some(Self::parse_buy_exact_out_instruction), }, GenericEventParseConfig { program_id: BONK_PROGRAM_ID, @@ -57,8 +57,8 @@ impl BonkEventParser { inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::SELL_EXACT_IN, event_type: EventType::BonkSellExactIn, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_sell_exact_in_instruction, + inner_instruction_parser: Some(Self::parse_trade_inner_instruction), + instruction_parser: Some(Self::parse_sell_exact_in_instruction), }, GenericEventParseConfig { program_id: BONK_PROGRAM_ID, @@ -66,8 +66,8 @@ impl BonkEventParser { inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::SELL_EXACT_OUT, event_type: EventType::BonkSellExactOut, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_sell_exact_out_instruction, + inner_instruction_parser: Some(Self::parse_trade_inner_instruction), + instruction_parser: Some(Self::parse_sell_exact_out_instruction), }, GenericEventParseConfig { program_id: BONK_PROGRAM_ID, @@ -75,8 +75,8 @@ impl BonkEventParser { inner_instruction_discriminator: discriminators::POOL_CREATE_EVENT, instruction_discriminator: discriminators::INITIALIZE, event_type: EventType::BonkInitialize, - inner_instruction_parser: Self::parse_pool_create_inner_instruction, - instruction_parser: Self::parse_initialize_instruction, + inner_instruction_parser: Some(Self::parse_pool_create_inner_instruction), + instruction_parser: Some(Self::parse_initialize_instruction), }, GenericEventParseConfig { program_id: BONK_PROGRAM_ID, @@ -84,8 +84,8 @@ impl BonkEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::MIGRATE_TO_AMM, event_type: EventType::BonkMigrateToAmm, - inner_instruction_parser: Self::parse_migrate_to_amm_inner_instruction, - instruction_parser: Self::parse_migrate_to_amm_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_migrate_to_amm_instruction), }, GenericEventParseConfig { program_id: BONK_PROGRAM_ID, @@ -93,8 +93,8 @@ impl BonkEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::MIGRATE_TO_CP_SWAP, event_type: EventType::BonkMigrateToCpswap, - inner_instruction_parser: Self::parse_migrate_to_cpswap_inner_instruction, - instruction_parser: Self::parse_migrate_to_cpswap_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_migrate_to_cpswap_instruction), }, ]; @@ -117,22 +117,6 @@ impl BonkEventParser { } } - /// Parse migrate to AMM event - fn parse_migrate_to_amm_inner_instruction( - _data: &[u8], - _metadata: EventMetadata, - ) -> Option> { - None - } - - /// Parse migrate to CP Swap event - fn parse_migrate_to_cpswap_inner_instruction( - _data: &[u8], - _metadata: EventMetadata, - ) -> Option> { - None - } - /// Parse trade event fn parse_trade_inner_instruction( data: &[u8], diff --git a/src/streaming/event_parser/protocols/bonk/types.rs b/src/streaming/event_parser/protocols/bonk/types.rs index 5751f8c..af9f576 100755 --- a/src/streaming/event_parser/protocols/bonk/types.rs +++ b/src/streaming/event_parser/protocols/bonk/types.rs @@ -1,5 +1,15 @@ use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::{ + event_parser::{ + common::EventMetadata, + protocols::bonk::{BonkGlobalConfigAccountEvent, BonkPlatformConfigAccountEvent, BonkPoolStateAccountEvent}, + UnifiedEvent, + }, + grpc::AccountPretty, +}; #[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub enum TradeDirection { @@ -62,8 +72,165 @@ pub enum CurveParams { impl Default for CurveParams { fn default() -> Self { - Self::Constant { - data: ConstantCurve::default(), - } + Self::Constant { data: ConstantCurve::default() } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct VestingSchedule { + pub total_locked_amount: u64, + pub cliff_period: u64, + pub unlock_period: u64, + pub start_time: u64, + pub allocated_share_amount: u64, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PoolState { + pub epoch: u64, + pub auth_bump: u8, + pub status: u8, + pub base_decimals: u8, + pub quote_decimals: u8, + pub migrate_type: u8, + pub supply: u64, + pub total_base_sell: u64, + pub virtual_base: u64, + pub virtual_quote: u64, + pub real_base: u64, + pub real_quote: u64, + pub total_quote_fund_raising: u64, + pub quote_protocol_fee: u64, + pub platform_fee: u64, + pub migrate_fee: u64, + pub vesting_schedule: VestingSchedule, + pub global_config: Pubkey, + pub platform_config: Pubkey, + pub base_mint: Pubkey, + pub quote_mint: Pubkey, + pub base_vault: Pubkey, + pub quote_vault: Pubkey, + pub creator: Pubkey, + pub padding: [u64; 8], +} + +pub const POOL_STATE_SIZE: usize = 8 + 1 * 5 + 8 * 10 + 32 * 7 + 8 * 8 + 8 * 5; + +pub fn pool_state_decode(data: &[u8]) -> Option { + if data.len() < POOL_STATE_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn pool_state_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { + Some(Box::new(BonkPoolStateAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + pool_state, + })) + } else { + None + } +} +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct GlobalConfig { + pub epoch: u64, + pub curve_type: u8, + pub index: u16, + pub migrate_fee: u64, + pub trade_fee_rate: u64, + pub max_share_fee_rate: u64, + pub min_base_supply: u64, + pub max_lock_rate: u64, + pub min_base_sell_rate: u64, + pub min_base_migrate_rate: u64, + pub min_quote_fund_raising: u64, + pub quote_mint: Pubkey, + pub protocol_fee_owner: Pubkey, + pub migrate_fee_owner: Pubkey, + pub migrate_to_amm_wallet: Pubkey, + pub migrate_to_cpswap_wallet: Pubkey, + pub padding: [u64; 16], +} + +pub const GLOBAL_CONFIG_SIZE: usize = 8 + 1 + 2 + 8 * 8 + 32 * 5 + 8 * 16; + +pub fn global_config_decode(data: &[u8]) -> Option { + if data.len() < GLOBAL_CONFIG_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn global_config_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(global_config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { + Some(Box::new(BonkGlobalConfigAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + global_config, + })) + } else { + None + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PlatformConfig { + pub epoch: u64, + pub platform_fee_wallet: Pubkey, + pub platform_nft_wallet: Pubkey, + pub platform_scale: u64, + pub creator_scale: u64, + pub burn_scale: u64, + pub fee_rate: u64, + pub name: Vec, + pub web: Vec, + pub img: Vec, + pub padding: Vec, +} + +pub const PLATFORM_CONFIG_SIZE: usize = 8 + 32 * 2 + 8 * 4 + 8 * 64 + 8 * 256 + 8 * 256 + 8 * 256; + +pub fn platform_config_decode(data: &[u8]) -> Option { + if data.len() < PLATFORM_CONFIG_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn platform_config_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(platform_config) = + platform_config_decode(&account.data[8..PLATFORM_CONFIG_SIZE + 8]) + { + Some(Box::new(BonkPlatformConfigAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + platform_config, + })) + } else { + None } } diff --git a/src/streaming/event_parser/protocols/pumpfun/events.rs b/src/streaming/event_parser/protocols/pumpfun/events.rs index 7f3354f..14ab0bc 100755 --- a/src/streaming/event_parser/protocols/pumpfun/events.rs +++ b/src/streaming/event_parser/protocols/pumpfun/events.rs @@ -4,6 +4,7 @@ use solana_sdk::pubkey::Pubkey; use crate::impl_unified_event; use crate::streaming::event_parser::common::EventMetadata; +use crate::streaming::event_parser::protocols::pumpfun::types::{BondingCurve, Global}; #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] pub struct PumpFunCreateTokenEvent { @@ -177,6 +178,35 @@ impl_unified_event!( pool ); +/// 铸币曲线 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PumpFunBondingCurveAccountEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub bonding_curve: BondingCurve, +} + +impl_unified_event!(PumpFunBondingCurveAccountEvent,); + +/// 全局配置 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PumpFunGlobalAccountEvent { + #[borsh(skip)] + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub global: Global, +} +impl_unified_event!(PumpFunGlobalAccountEvent,); + /// 事件鉴别器常量 pub mod discriminators { // 事件鉴别器 @@ -189,4 +219,8 @@ pub mod discriminators { pub const BUY_IX: &[u8] = &[102, 6, 61, 18, 1, 218, 235, 234]; pub const SELL_IX: &[u8] = &[51, 230, 133, 164, 1, 127, 131, 173]; pub const MIGRATE_IX: &[u8] = &[155, 234, 231, 146, 236, 158, 162, 30]; + + // 账户鉴别器 + pub const BONDING_CURVE_ACCOUNT: &[u8] = &[23, 183, 248, 55, 96, 216, 172, 96]; + pub const GLOBAL_ACCOUNT: &[u8] = &[167, 232, 232, 177, 200, 108, 114, 127]; } diff --git a/src/streaming/event_parser/protocols/pumpfun/mod.rs b/src/streaming/event_parser/protocols/pumpfun/mod.rs index eee7acf..fb51532 100755 --- a/src/streaming/event_parser/protocols/pumpfun/mod.rs +++ b/src/streaming/event_parser/protocols/pumpfun/mod.rs @@ -1,5 +1,6 @@ pub mod events; pub mod parser; +pub mod types; pub use events::*; pub use parser::PumpFunEventParser; \ No newline at end of file diff --git a/src/streaming/event_parser/protocols/pumpfun/parser.rs b/src/streaming/event_parser/protocols/pumpfun/parser.rs index 02ae8cc..2589897 100755 --- a/src/streaming/event_parser/protocols/pumpfun/parser.rs +++ b/src/streaming/event_parser/protocols/pumpfun/parser.rs @@ -37,8 +37,8 @@ impl PumpFunEventParser { inner_instruction_discriminator: discriminators::CREATE_TOKEN_EVENT, instruction_discriminator: discriminators::CREATE_TOKEN_IX, event_type: EventType::PumpFunCreateToken, - inner_instruction_parser: Self::parse_create_token_inner_instruction, - instruction_parser: Self::parse_create_token_instruction, + inner_instruction_parser: Some(Self::parse_create_token_inner_instruction), + instruction_parser: Some(Self::parse_create_token_instruction), }, GenericEventParseConfig { program_id: PUMPFUN_PROGRAM_ID, @@ -46,8 +46,8 @@ impl PumpFunEventParser { inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::BUY_IX, event_type: EventType::PumpFunBuy, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_buy_instruction, + inner_instruction_parser: Some(Self::parse_trade_inner_instruction), + instruction_parser: Some(Self::parse_buy_instruction), }, GenericEventParseConfig { program_id: PUMPFUN_PROGRAM_ID, @@ -55,8 +55,8 @@ impl PumpFunEventParser { inner_instruction_discriminator: discriminators::TRADE_EVENT, instruction_discriminator: discriminators::SELL_IX, event_type: EventType::PumpFunSell, - inner_instruction_parser: Self::parse_trade_inner_instruction, - instruction_parser: Self::parse_sell_instruction, + inner_instruction_parser: Some(Self::parse_trade_inner_instruction), + instruction_parser: Some(Self::parse_sell_instruction), }, GenericEventParseConfig { program_id: PUMPFUN_PROGRAM_ID, @@ -64,8 +64,8 @@ impl PumpFunEventParser { inner_instruction_discriminator: discriminators::COMPLETE_PUMP_AMM_MIGRATION_EVENT, instruction_discriminator: discriminators::MIGRATE_IX, event_type: EventType::PumpFunMigrate, - inner_instruction_parser: Self::parse_migrate_inner_instruction, - instruction_parser: Self::parse_migrate_instruction, + inner_instruction_parser: Some(Self::parse_migrate_inner_instruction), + instruction_parser: Some(Self::parse_migrate_instruction), }, ]; diff --git a/src/streaming/event_parser/protocols/pumpfun/types.rs b/src/streaming/event_parser/protocols/pumpfun/types.rs new file mode 100644 index 0000000..cdeeb7f --- /dev/null +++ b/src/streaming/event_parser/protocols/pumpfun/types.rs @@ -0,0 +1,98 @@ +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::{ + event_parser::{ + common::EventMetadata, + protocols::pumpfun::{PumpFunBondingCurveAccountEvent, PumpFunGlobalAccountEvent}, + UnifiedEvent, + }, + grpc::AccountPretty, +}; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct BondingCurve { + pub virtual_token_reserves: u64, + pub virtual_sol_reserves: u64, + pub real_token_reserves: u64, + pub real_sol_reserves: u64, + pub token_total_supply: u64, + pub complete: bool, + pub creator: Pubkey, +} + +pub const BONDING_CURVE_SIZE: usize = 8 * 5 + 1 + 32; + +pub fn bonding_curve_decode(data: &[u8]) -> Option { + if data.len() < BONDING_CURVE_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn bonding_curve_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(bonding_curve) = bonding_curve_decode(&account.data[8..BONDING_CURVE_SIZE + 8]) { + Some(Box::new(PumpFunBondingCurveAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + bonding_curve, + })) + } else { + None + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct Global { + pub initialized: bool, + pub authority: Pubkey, + pub fee_recipient: Pubkey, + pub initial_virtual_token_reserves: u64, + pub initial_virtual_sol_reserves: u64, + pub initial_real_token_reserves: u64, + pub token_total_supply: u64, + pub fee_basis_points: u64, + pub withdraw_authority: Pubkey, + pub enable_migrate: bool, + pub pool_migration_fee: u64, + pub creator_fee_basis_points: u64, + pub fee_recipients: [Pubkey; 7], + pub set_creator_authority: Pubkey, + pub admin_set_creator_authority: Pubkey, +} + +pub const GLOBAL_SIZE: usize = 1 + 32 * 2 + 8 * 5 + 32 + 1 + 8 * 2 + 32 * 7 + 32 * 2; + +pub fn global_decode(data: &[u8]) -> Option { + if data.len() < GLOBAL_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn global_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(global) = global_decode(&account.data[8..GLOBAL_SIZE + 8]) { + Some(Box::new(PumpFunGlobalAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + global, + })) + } else { + None + } +} diff --git a/src/streaming/event_parser/protocols/pumpswap/events.rs b/src/streaming/event_parser/protocols/pumpswap/events.rs index 64b649c..215835f 100755 --- a/src/streaming/event_parser/protocols/pumpswap/events.rs +++ b/src/streaming/event_parser/protocols/pumpswap/events.rs @@ -2,8 +2,9 @@ use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; -use crate::streaming::event_parser::common::EventMetadata; use crate::impl_unified_event; +use crate::streaming::event_parser::common::EventMetadata; +use crate::streaming::event_parser::protocols::pumpswap::types::{GlobalConfig, Pool}; /// 买入事件 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] @@ -309,6 +310,32 @@ impl_unified_event!( user_pool_token_account ); +/// 全局配置 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PumpSwapGlobalConfigAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub global_config: GlobalConfig, +} +impl_unified_event!(PumpSwapGlobalConfigAccountEvent,); + +/// 池 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PumpSwapPoolAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub pool: Pool, +} +impl_unified_event!(PumpSwapPoolAccountEvent,); + /// 事件鉴别器常量 pub mod discriminators { // 事件鉴别器 @@ -324,4 +351,8 @@ pub mod discriminators { pub const CREATE_POOL_IX: &[u8] = &[233, 146, 209, 142, 207, 104, 64, 188]; pub const DEPOSIT_IX: &[u8] = &[242, 35, 198, 137, 82, 225, 242, 182]; pub const WITHDRAW_IX: &[u8] = &[183, 18, 70, 156, 148, 109, 161, 34]; + + // 账户鉴别器 + pub const GLOBAL_CONFIG_ACCOUNT: &[u8] = &[149, 8, 156, 202, 160, 252, 176, 217]; + pub const POOL_ACCOUNT: &[u8] = &[241, 154, 109, 4, 17, 177, 109, 188]; } diff --git a/src/streaming/event_parser/protocols/pumpswap/mod.rs b/src/streaming/event_parser/protocols/pumpswap/mod.rs index cfa440f..ee80dba 100755 --- a/src/streaming/event_parser/protocols/pumpswap/mod.rs +++ b/src/streaming/event_parser/protocols/pumpswap/mod.rs @@ -1,5 +1,6 @@ pub mod events; pub mod parser; +pub mod types; pub use events::*; -pub use parser::PumpSwapEventParser; \ No newline at end of file +pub use parser::PumpSwapEventParser; diff --git a/src/streaming/event_parser/protocols/pumpswap/parser.rs b/src/streaming/event_parser/protocols/pumpswap/parser.rs index ce59bb3..c55be69 100755 --- a/src/streaming/event_parser/protocols/pumpswap/parser.rs +++ b/src/streaming/event_parser/protocols/pumpswap/parser.rs @@ -38,8 +38,8 @@ impl PumpSwapEventParser { inner_instruction_discriminator: discriminators::BUY_EVENT, instruction_discriminator: discriminators::BUY_IX, event_type: EventType::PumpSwapBuy, - inner_instruction_parser: Self::parse_buy_inner_instruction, - instruction_parser: Self::parse_buy_instruction, + inner_instruction_parser: Some(Self::parse_buy_inner_instruction), + instruction_parser: Some(Self::parse_buy_instruction), }, GenericEventParseConfig { program_id: PUMPSWAP_PROGRAM_ID, @@ -47,8 +47,8 @@ impl PumpSwapEventParser { inner_instruction_discriminator: discriminators::SELL_EVENT, instruction_discriminator: discriminators::SELL_IX, event_type: EventType::PumpSwapSell, - inner_instruction_parser: Self::parse_sell_inner_instruction, - instruction_parser: Self::parse_sell_instruction, + inner_instruction_parser: Some(Self::parse_sell_inner_instruction), + instruction_parser: Some(Self::parse_sell_instruction), }, GenericEventParseConfig { program_id: PUMPSWAP_PROGRAM_ID, @@ -56,8 +56,8 @@ impl PumpSwapEventParser { inner_instruction_discriminator: discriminators::CREATE_POOL_EVENT, instruction_discriminator: discriminators::CREATE_POOL_IX, event_type: EventType::PumpSwapCreatePool, - inner_instruction_parser: Self::parse_create_pool_inner_instruction, - instruction_parser: Self::parse_create_pool_instruction, + inner_instruction_parser: Some(Self::parse_create_pool_inner_instruction), + instruction_parser: Some(Self::parse_create_pool_instruction), }, GenericEventParseConfig { program_id: PUMPSWAP_PROGRAM_ID, @@ -65,8 +65,8 @@ impl PumpSwapEventParser { inner_instruction_discriminator: discriminators::DEPOSIT_EVENT, instruction_discriminator: discriminators::DEPOSIT_IX, event_type: EventType::PumpSwapDeposit, - inner_instruction_parser: Self::parse_deposit_inner_instruction, - instruction_parser: Self::parse_deposit_instruction, + inner_instruction_parser: Some(Self::parse_deposit_inner_instruction), + instruction_parser: Some(Self::parse_deposit_instruction), }, GenericEventParseConfig { program_id: PUMPSWAP_PROGRAM_ID, @@ -74,8 +74,8 @@ impl PumpSwapEventParser { inner_instruction_discriminator: discriminators::WITHDRAW_EVENT, instruction_discriminator: discriminators::WITHDRAW_IX, event_type: EventType::PumpSwapWithdraw, - inner_instruction_parser: Self::parse_withdraw_inner_instruction, - instruction_parser: Self::parse_withdraw_instruction, + inner_instruction_parser: Some(Self::parse_withdraw_inner_instruction), + instruction_parser: Some(Self::parse_withdraw_instruction), }, ]; diff --git a/src/streaming/event_parser/protocols/pumpswap/types.rs b/src/streaming/event_parser/protocols/pumpswap/types.rs new file mode 100644 index 0000000..0020dc6 --- /dev/null +++ b/src/streaming/event_parser/protocols/pumpswap/types.rs @@ -0,0 +1,95 @@ +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::{ + event_parser::{ + common::EventMetadata, + protocols::pumpswap::{ + parser::PUMPSWAP_PROGRAM_ID, PumpSwapGlobalConfigAccountEvent, PumpSwapPoolAccountEvent, + }, + UnifiedEvent, + }, + grpc::AccountPretty, +}; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct GlobalConfig { + pub admin: Pubkey, + pub lp_fee_basis_points: u64, + pub protocol_fee_basis_points: u64, + pub disable_flags: u8, + pub protocol_fee_recipients: [Pubkey; 8], + pub coin_creator_fee_basis_points: u64, + pub admin_set_coin_creator_authority: Pubkey, +} + +pub const GLOBAL_CONFIG_SIZE: usize = 32 + 8 + 8 + 1 + 32 * 8 + 8 + 32; + +pub fn global_config_decode(data: &[u8]) -> Option { + if data.len() < GLOBAL_CONFIG_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn global_config_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) { + Some(Box::new(PumpSwapGlobalConfigAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + global_config: config, + })) + } else { + None + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct Pool { + pub pool_bump: u8, + pub index: u16, + pub creator: Pubkey, + pub base_mint: Pubkey, + pub quote_mint: Pubkey, + pub lp_mint: Pubkey, + pub pool_base_token_account: Pubkey, + pub pool_quote_token_account: Pubkey, + pub lp_supply: u64, + pub coin_creator: Pubkey, +} + +pub const POOL_SIZE: usize = 1 + 2 + 32 * 6 + 8 + 32; + +pub fn pool_decode(data: &[u8]) -> Option { + if data.len() < POOL_SIZE { + return None; + } + borsh::from_slice::(&data).ok() +} + +pub fn pool_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if let Some(pool) = pool_decode(&account.data[8..POOL_SIZE + 8]) { + Some(Box::new(PumpSwapPoolAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + pool: pool, + })) + } else { + None + } +} diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs index d68f495..cd4eb19 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/events.rs @@ -1,5 +1,7 @@ -use crate::impl_unified_event; use crate::streaming::event_parser::common::EventMetadata; +use crate::{ + impl_unified_event, streaming::event_parser::protocols::raydium_amm_v4::types::AmmInfo, +}; use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; @@ -151,6 +153,19 @@ pub struct RaydiumAmmV4WithdrawPnlEvent { } impl_unified_event!(RaydiumAmmV4WithdrawPnlEvent,); +/// 池信息 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct RaydiumAmmV4AmmInfoAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub amm_info: AmmInfo, +} +impl_unified_event!(RaydiumAmmV4AmmInfoAccountEvent,); + /// 事件鉴别器常量 pub mod discriminators { // 指令鉴别器 @@ -160,4 +175,7 @@ pub mod discriminators { pub const INITIALIZE2: &[u8] = &[01]; pub const WITHDRAW: &[u8] = &[04]; pub const WITHDRAW_PNL: &[u8] = &[07]; + + /// 池信息鉴别器 + pub const AMM_INFO: &[u8] = &[6]; } diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/mod.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/mod.rs index 552d4af..b8e0ce5 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/mod.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/mod.rs @@ -1,5 +1,6 @@ pub mod events; pub mod parser; +pub mod types; pub use events::*; pub use parser::RaydiumAmmV4EventParser; diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs index adbe808..ba4a5cf 100755 --- a/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/parser.rs @@ -38,8 +38,8 @@ impl RaydiumAmmV4EventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_BASE_IN, event_type: EventType::RaydiumAmmV4SwapBaseIn, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_swap_base_input_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_swap_base_input_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_AMM_V4_PROGRAM_ID, @@ -47,8 +47,8 @@ impl RaydiumAmmV4EventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_BASE_OUT, event_type: EventType::RaydiumAmmV4SwapBaseOut, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_swap_base_output_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_swap_base_output_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_AMM_V4_PROGRAM_ID, @@ -56,8 +56,8 @@ impl RaydiumAmmV4EventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::DEPOSIT, event_type: EventType::RaydiumAmmV4Deposit, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_deposit_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_deposit_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_AMM_V4_PROGRAM_ID, @@ -65,8 +65,8 @@ impl RaydiumAmmV4EventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::INITIALIZE2, event_type: EventType::RaydiumAmmV4Initialize2, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_initialize2_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_initialize2_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_AMM_V4_PROGRAM_ID, @@ -74,8 +74,8 @@ impl RaydiumAmmV4EventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::WITHDRAW, event_type: EventType::RaydiumAmmV4Withdraw, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_withdraw_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_withdraw_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_AMM_V4_PROGRAM_ID, @@ -83,8 +83,8 @@ impl RaydiumAmmV4EventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::WITHDRAW_PNL, event_type: EventType::RaydiumAmmV4WithdrawPnl, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_withdraw_pnl_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_withdraw_pnl_instruction), }, ]; @@ -93,10 +93,6 @@ impl RaydiumAmmV4EventParser { Self { inner } } - fn empty_parse(_data: &[u8], _metadata: EventMetadata) -> Option> { - None - } - /// 解析提现指令事件 fn parse_withdraw_pnl_instruction( _data: &[u8], diff --git a/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs new file mode 100644 index 0000000..f7c2117 --- /dev/null +++ b/src/streaming/event_parser/protocols/raydium_amm_v4/types.rs @@ -0,0 +1,106 @@ +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::{ + event_parser::{ + common::EventMetadata, protocols::raydium_amm_v4::RaydiumAmmV4AmmInfoAccountEvent, + UnifiedEvent, + }, + grpc::AccountPretty, +}; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct Fees { + pub min_separate_numerator: u64, + pub min_separate_denominator: u64, + pub trade_fee_numerator: u64, + pub trade_fee_denominator: u64, + pub pnl_numerator: u64, + pub pnl_denominator: u64, + pub swap_fee_numerator: u64, + pub swap_fee_denominator: u64, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct OutPutData { + pub need_take_pnl_coin: u64, + pub need_take_pnl_pc: u64, + pub total_pnl_pc: u64, + pub total_pnl_coin: u64, + pub pool_open_time: u64, + pub punish_pc_amount: u64, + pub punish_coin_amount: u64, + pub orderbook_to_init_time: u64, + pub swap_coin_in_amount: u128, + pub swap_pc_out_amount: u128, + pub swap_take_pc_fee: u64, + pub swap_pc_in_amount: u128, + pub swap_coin_out_amount: u128, + pub swap_take_coin_fee: u64, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct AmmInfo { + pub status: u64, + pub nonce: u64, + pub order_num: u64, + pub depth: u64, + pub coin_decimals: u64, + pub pc_decimals: u64, + pub state: u64, + pub reset_flag: u64, + pub min_size: u64, + pub vol_max_cut_ratio: u64, + pub amount_wave: u64, + pub coin_lot_size: u64, + pub pc_lot_size: u64, + pub min_price_multiplier: u64, + pub max_price_multiplier: u64, + pub sys_decimal_value: u64, + pub fees: Fees, + pub out_put: OutPutData, + pub token_coin: Pubkey, + pub token_pc: Pubkey, + pub coin_mint: Pubkey, + pub pc_mint: Pubkey, + pub lp_mint: Pubkey, + pub open_orders: Pubkey, + pub market: Pubkey, + pub serum_dex: Pubkey, + pub target_orders: Pubkey, + pub withdraw_queue: Pubkey, + pub token_temp_lp: Pubkey, + pub amm_owner: Pubkey, + pub lp_amount: u64, + pub client_order_id: u64, + pub padding: [u64; 2], +} + +pub const AMM_INFO_SIZE: usize = 752; + +pub fn amm_info_decode(data: &[u8]) -> Option { + borsh::from_slice::(&data).ok() +} + +pub fn amm_info_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if account.data.len() < AMM_INFO_SIZE { + return None; + } + if let Some(amm_info) = amm_info_decode(&account.data[..AMM_INFO_SIZE]) { + Some(Box::new(RaydiumAmmV4AmmInfoAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + amm_info: amm_info, + })) + } else { + None + } +} diff --git a/src/streaming/event_parser/protocols/raydium_clmm/events.rs b/src/streaming/event_parser/protocols/raydium_clmm/events.rs index 6add233..f8834fa 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/events.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/events.rs @@ -1,6 +1,8 @@ -use crate::impl_unified_event; use crate::streaming::event_parser::common::EventMetadata; -// use borsh::BorshDeserialize; +use crate::streaming::event_parser::protocols::raydium_clmm::types::{PoolState, TickArrayState}; +use crate::{ + impl_unified_event, streaming::event_parser::protocols::raydium_clmm::types::AmmConfig, +}; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; @@ -217,6 +219,45 @@ pub struct RaydiumClmmOpenPositionV2Event { } impl_unified_event!(RaydiumClmmOpenPositionV2Event,); +/// 池配置 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RaydiumClmmAmmConfigAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub amm_config: AmmConfig, +} +impl_unified_event!(RaydiumClmmAmmConfigAccountEvent,); + +/// 池状态 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RaydiumClmmPoolStateAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub pool_state: PoolState, +} +impl_unified_event!(RaydiumClmmPoolStateAccountEvent,); + +/// 池状态 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RaydiumClmmTickArrayStateAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub tick_array_state: TickArrayState, +} +impl_unified_event!(RaydiumClmmTickArrayStateAccountEvent,); + /// 事件鉴别器常量 pub mod discriminators { // 指令鉴别器 @@ -228,4 +269,9 @@ pub mod discriminators { pub const CREATE_POOL: &[u8] = &[233, 146, 209, 142, 207, 104, 64, 188]; pub const OPEN_POSITION_WITH_TOKEN_22_NFT: &[u8] = &[77, 255, 174, 82, 125, 29, 201, 46]; pub const OPEN_POSITION_V2: &[u8] = &[77, 184, 74, 214, 112, 86, 241, 199]; + + // 账号鉴别器 + pub const AMM_CONFIG: &[u8] = &[218, 244, 33, 104, 203, 203, 43, 111]; + pub const POOL_STATE: &[u8] = &[247, 237, 227, 245, 215, 195, 222, 70]; + pub const TICK_ARRAY_STATE: &[u8] = &[192, 155, 85, 205, 49, 249, 129, 42]; } diff --git a/src/streaming/event_parser/protocols/raydium_clmm/mod.rs b/src/streaming/event_parser/protocols/raydium_clmm/mod.rs index 89b867a..c5eba4c 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/mod.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/mod.rs @@ -1,5 +1,6 @@ pub mod events; pub mod parser; +pub mod types; pub use events::*; pub use parser::RaydiumClmmEventParser; diff --git a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs index e5777ce..f06f238 100755 --- a/src/streaming/event_parser/protocols/raydium_clmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_clmm/parser.rs @@ -43,8 +43,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP, event_type: EventType::RaydiumClmmSwap, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_swap_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_swap_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -52,8 +52,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_V2, event_type: EventType::RaydiumClmmSwapV2, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_swap_v2_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_swap_v2_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -61,8 +61,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::CLOSE_POSITION, event_type: EventType::RaydiumClmmClosePosition, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_close_position_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_close_position_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -70,8 +70,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::DECREASE_LIQUIDITY_V2, event_type: EventType::RaydiumClmmDecreaseLiquidityV2, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_decrease_liquidity_v2_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_decrease_liquidity_v2_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -79,8 +79,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::CREATE_POOL, event_type: EventType::RaydiumClmmCreatePool, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_create_pool_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_create_pool_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -88,8 +88,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::INCREASE_LIQUIDITY_V2, event_type: EventType::RaydiumClmmIncreaseLiquidityV2, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_increase_liquidity_v2_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_increase_liquidity_v2_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -97,8 +97,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::OPEN_POSITION_WITH_TOKEN_22_NFT, event_type: EventType::RaydiumClmmOpenPositionWithToken22Nft, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_open_position_with_token_22_nft_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_open_position_with_token_22_nft_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CLMM_PROGRAM_ID, @@ -106,8 +106,8 @@ impl RaydiumClmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::OPEN_POSITION_V2, event_type: EventType::RaydiumClmmOpenPositionV2, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_open_position_v2_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_open_position_v2_instruction), }, ]; @@ -116,10 +116,6 @@ impl RaydiumClmmEventParser { Self { inner } } - fn empty_parse(_data: &[u8], _metadata: EventMetadata) -> Option> { - None - } - /// 解析打开仓位V2指令事件 fn parse_open_position_v2_instruction( data: &[u8], diff --git a/src/streaming/event_parser/protocols/raydium_clmm/types.rs b/src/streaming/event_parser/protocols/raydium_clmm/types.rs new file mode 100644 index 0000000..39834ca --- /dev/null +++ b/src/streaming/event_parser/protocols/raydium_clmm/types.rs @@ -0,0 +1,222 @@ +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::{ + event_parser::{ + common::EventMetadata, + protocols::raydium_clmm::{ + RaydiumClmmAmmConfigAccountEvent, RaydiumClmmPoolStateAccountEvent, + RaydiumClmmTickArrayStateAccountEvent, + }, + UnifiedEvent, + }, + grpc::AccountPretty, +}; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct AmmConfig { + pub bump: u8, + pub index: u16, + pub owner: Pubkey, + pub protocol_fee_rate: u32, + pub trade_fee_rate: u32, + pub tick_spacing: u16, + pub fund_fee_rate: u32, + pub padding_u32: u32, + pub fund_owner: Pubkey, + pub padding: [u64; 3], +} + +pub const AMM_CONFIG_SIZE: usize = 1 + 2 + 32 + 4 * 2 + 2 + 4 * 2 + 32 + 8 * 3; + +pub fn amm_config_decode(data: &[u8]) -> Option { + borsh::from_slice::(&data).ok() +} + +pub fn amm_config_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if account.data.len() < AMM_CONFIG_SIZE { + return None; + } + if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { + Some(Box::new(RaydiumClmmAmmConfigAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + amm_config: amm_config, + })) + } else { + None + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct RewardInfo { + pub reward_state: u8, + pub open_time: u64, + pub end_time: u64, + pub last_update_time: u64, + pub emissions_per_second_x64: u128, + pub reward_total_emissioned: u64, + pub reward_claimed: u64, + pub token_mint: Pubkey, + pub token_vault: Pubkey, + pub authority: Pubkey, + pub reward_growth_global_x64: u128, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PoolState { + pub bump: [u8; 1], + pub amm_config: Pubkey, + pub owner: Pubkey, + pub token_mint0: Pubkey, + pub token_mint1: Pubkey, + pub token_vault0: Pubkey, + pub token_vault1: Pubkey, + pub observation_key: Pubkey, + pub mint_decimals0: u8, + pub mint_decimals1: u8, + pub tick_spacing: u16, + pub liquidity: u128, + pub sqrt_price_x64: u128, + pub tick_current: i32, + pub padding3: u16, + pub padding4: u16, + pub fee_growth_global0_x64: u128, + pub fee_growth_global1_x64: u128, + pub protocol_fees_token0: u64, + pub protocol_fees_token1: u64, + pub swap_in_amount_token0: u128, + pub swap_out_amount_token1: u128, + pub swap_in_amount_token1: u128, + pub swap_out_amount_token0: u128, + pub status: u8, + pub padding: [u8; 7], + pub reward_infos: [RewardInfo; 3], + pub tick_array_bitmap: [u64; 16], + pub total_fees_token0: u64, + pub total_fees_claimed_token0: u64, + pub total_fees_token1: u64, + pub total_fees_claimed_token1: u64, + pub fund_fees_token0: u64, + pub fund_fees_token1: u64, + pub open_time: u64, + pub recent_epoch: u64, + pub padding1: [u64; 24], + pub padding2: [u64; 32], +} + +pub const POOL_STATE_SIZE: usize = 1536; + +pub fn pool_state_decode(data: &[u8]) -> Option { + borsh::from_slice::(&data).ok() +} + +pub fn pool_state_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if account.data.len() < POOL_STATE_SIZE { + return None; + } + if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { + Some(Box::new(RaydiumClmmPoolStateAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + pool_state: pool_state, + })) + } else { + None + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct TickState { + pub tick: i32, + pub liquidity_net: i128, + pub liquidity_gross: u128, + pub fee_growth_outside0_x64: u128, + pub fee_growth_outside1_x64: u128, + pub reward_growths_outside_x64: [u128; 3], + pub padding: [u32; 13], +} + +impl Default for TickState { + fn default() -> Self { + Self { + tick: 0, + liquidity_net: 0, + liquidity_gross: 0, + fee_growth_outside0_x64: 0, + fee_growth_outside1_x64: 0, + reward_growths_outside_x64: [0; 3], + padding: [0; 13], + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct TickArrayState { + pub pool_id: Pubkey, + pub start_tick_index: i32, + #[serde(with = "serde_big_array::BigArray")] + pub ticks: [TickState; 60], + pub initialized_tick_count: u8, + pub recent_epoch: u64, + #[serde(with = "serde_big_array::BigArray")] + pub padding: [u8; 107], +} + +impl Default for TickArrayState { + fn default() -> Self { + Self { + pool_id: Pubkey::default(), + start_tick_index: 0, + ticks: core::array::from_fn(|_| TickState::default()), + initialized_tick_count: 0, + recent_epoch: 0, + padding: [0u8; 107], + } + } +} + +pub const TICK_ARRAY_STATE_SIZE: usize = 10232; + +pub fn tick_array_state_decode(data: &[u8]) -> Option { + borsh::from_slice::(&data).ok() +} + +pub fn tick_array_state_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if account.data.len() < TICK_ARRAY_STATE_SIZE { + return None; + } + if let Some(tick_array_state) = + tick_array_state_decode(&account.data[8..TICK_ARRAY_STATE_SIZE + 8]) + { + Some(Box::new(RaydiumClmmTickArrayStateAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + tick_array_state: tick_array_state, + })) + } else { + None + } +} diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs index 92cc101..39f8957 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/events.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/events.rs @@ -1,5 +1,8 @@ -use crate::impl_unified_event; use crate::streaming::event_parser::common::EventMetadata; +use crate::streaming::event_parser::protocols::raydium_cpmm::types::PoolState; +use crate::{ + impl_unified_event, streaming::event_parser::protocols::raydium_cpmm::types::AmmConfig, +}; use borsh::BorshDeserialize; use serde::{Deserialize, Serialize}; use solana_sdk::pubkey::Pubkey; @@ -107,6 +110,32 @@ pub struct RaydiumCpmmWithdrawEvent { } impl_unified_event!(RaydiumCpmmWithdrawEvent,); +/// 池配置 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct RaydiumCpmmAmmConfigAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub amm_config: AmmConfig, +} +impl_unified_event!(RaydiumCpmmAmmConfigAccountEvent,); + +/// 池状态 +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct RaydiumCpmmPoolStateAccountEvent { + pub metadata: EventMetadata, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub pool_state: PoolState, +} +impl_unified_event!(RaydiumCpmmPoolStateAccountEvent,); + /// 事件鉴别器常量 pub mod discriminators { // 指令鉴别器 @@ -115,4 +144,8 @@ pub mod discriminators { pub const DEPOSIT: &[u8] = &[242, 35, 198, 137, 82, 225, 242, 182]; pub const INITIALIZE: &[u8] = &[175, 175, 109, 31, 13, 152, 155, 237]; pub const WITHDRAW: &[u8] = &[183, 18, 70, 156, 148, 109, 161, 34]; + + // 账号鉴别器 + pub const AMM_CONFIG: &[u8] = &[218, 244, 33, 104, 203, 203, 43, 111]; + pub const POOL_STATE: &[u8] = &[247, 237, 227, 245, 215, 195, 222, 70]; } diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs b/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs index e5896a6..d24db63 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/mod.rs @@ -1,5 +1,6 @@ pub mod events; pub mod parser; +pub mod types; pub use events::*; pub use parser::RaydiumCpmmEventParser; diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs index a543685..6d08039 100755 --- a/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs +++ b/src/streaming/event_parser/protocols/raydium_cpmm/parser.rs @@ -38,8 +38,8 @@ impl RaydiumCpmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_BASE_IN, event_type: EventType::RaydiumCpmmSwapBaseInput, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_swap_base_input_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_swap_base_input_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CPMM_PROGRAM_ID, @@ -47,8 +47,8 @@ impl RaydiumCpmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::SWAP_BASE_OUT, event_type: EventType::RaydiumCpmmSwapBaseOutput, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_swap_base_output_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_swap_base_output_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CPMM_PROGRAM_ID, @@ -56,8 +56,8 @@ impl RaydiumCpmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::DEPOSIT, event_type: EventType::RaydiumCpmmDeposit, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_deposit_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_deposit_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CPMM_PROGRAM_ID, @@ -65,8 +65,8 @@ impl RaydiumCpmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::INITIALIZE, event_type: EventType::RaydiumCpmmInitialize, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_initialize_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_initialize_instruction), }, GenericEventParseConfig { program_id: RAYDIUM_CPMM_PROGRAM_ID, @@ -74,8 +74,8 @@ impl RaydiumCpmmEventParser { inner_instruction_discriminator: "", instruction_discriminator: discriminators::WITHDRAW, event_type: EventType::RaydiumCpmmWithdraw, - inner_instruction_parser: Self::empty_parse, - instruction_parser: Self::parse_withdraw_instruction, + inner_instruction_parser: None, + instruction_parser: Some(Self::parse_withdraw_instruction), }, ]; @@ -84,10 +84,6 @@ impl RaydiumCpmmEventParser { Self { inner } } - fn empty_parse(_data: &[u8], _metadata: EventMetadata) -> Option> { - None - } - /// 解析提款指令事件 fn parse_withdraw_instruction( data: &[u8], diff --git a/src/streaming/event_parser/protocols/raydium_cpmm/types.rs b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs new file mode 100644 index 0000000..eae06ff --- /dev/null +++ b/src/streaming/event_parser/protocols/raydium_cpmm/types.rs @@ -0,0 +1,111 @@ +use borsh::BorshDeserialize; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::streaming::{ + event_parser::{ + common::EventMetadata, + protocols::raydium_cpmm::{ + RaydiumCpmmAmmConfigAccountEvent, RaydiumCpmmPoolStateAccountEvent, + }, + UnifiedEvent, + }, + grpc::AccountPretty, +}; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct AmmConfig { + pub bump: u8, + pub disable_create_pool: bool, + pub index: u16, + pub trade_fee_rate: u64, + pub protocol_fee_rate: u64, + pub fund_fee_rate: u64, + pub create_pool_fee: u64, + pub protocol_owner: Pubkey, + pub fund_owner: Pubkey, + pub padding: [u64; 16], +} + +pub const AMM_CONFIG_SIZE: usize = 228; + +pub fn amm_config_decode(data: &[u8]) -> Option { + borsh::from_slice::(&data).ok() +} + +pub fn amm_config_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if account.data.len() < AMM_CONFIG_SIZE { + return None; + } + if let Some(amm_config) = amm_config_decode(&account.data[8..AMM_CONFIG_SIZE + 8]) { + Some(Box::new(RaydiumCpmmAmmConfigAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + amm_config: amm_config, + })) + } else { + None + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct PoolState { + pub amm_config: Pubkey, + pub pool_creator: Pubkey, + pub token0_vault: Pubkey, + pub token1_vault: Pubkey, + pub lp_mint: Pubkey, + pub token0_mint: Pubkey, + pub token1_mint: Pubkey, + pub token0_program: Pubkey, + pub token1_program: Pubkey, + pub observation_key: Pubkey, + pub auth_bump: u8, + pub status: u8, + pub lp_mint_decimals: u8, + pub mint0_decimals: u8, + pub mint1_decimals: u8, + pub lp_supply: u64, + pub protocol_fees_token0: u64, + pub protocol_fees_token1: u64, + pub fund_fees_token0: u64, + pub fund_fees_token1: u64, + pub open_time: u64, + pub recent_epoch: u64, + pub padding: [u64; 31], +} + +pub const POOL_STATE_SIZE: usize = 629; + +pub fn pool_state_decode(data: &[u8]) -> Option { + borsh::from_slice::(&data).ok() +} + +pub fn pool_state_parser( + account: &AccountPretty, + metadata: EventMetadata, +) -> Option> { + if account.data.len() < POOL_STATE_SIZE { + return None; + } + if let Some(pool_state) = pool_state_decode(&account.data[8..POOL_STATE_SIZE + 8]) { + Some(Box::new(RaydiumCpmmPoolStateAccountEvent { + metadata, + pubkey: account.pubkey.to_string(), + executable: account.executable, + lamports: account.lamports, + owner: account.owner.to_string(), + rent_epoch: account.rent_epoch, + pool_state: pool_state, + })) + } else { + None + } +} diff --git a/src/streaming/grpc/event_processor.rs b/src/streaming/grpc/event_processor.rs index 0780bf3..a315340 100644 --- a/src/streaming/grpc/event_processor.rs +++ b/src/streaming/grpc/event_processor.rs @@ -7,6 +7,7 @@ use crate::common::AnyResult; use crate::streaming::common::{ EventBatchProcessor as EventBatchCollector, MetricsManager, StreamClientConfig as ClientConfig, }; +use crate::streaming::event_parser::core::account_event_parser::AccountEventParser; use crate::streaming::event_parser::core::common_event_parser::CommonEventParser; use crate::streaming::event_parser::EventParser; use crate::streaming::event_parser::{ @@ -37,7 +38,28 @@ impl EventProcessor { F: Fn(Box) + Send + Sync, { match event_pretty { + EventPretty::Account(account_pretty) => { + self.metrics_manager.add_process_count().await; + let start_time = std::time::Instant::now(); + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); + let account_event = AccountEventParser::parse_account_event( + protocols.clone(), + account_pretty, + program_received_time_ms, + ); + if let Some(event) = account_event { + callback(event); + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + // 更新性能指标（如果启用） + self.metrics_manager.update_metrics(1, processing_time_ms).await; + // 记录慢处理操作 + self.metrics_manager.log_slow_processing(processing_time_ms, 1); + } + } EventPretty::Transaction(transaction_pretty) => { + self.metrics_manager.add_process_count().await; let start_time = std::time::Instant::now(); let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let slot = transaction_pretty.slot; @@ -76,16 +98,13 @@ impl EventProcessor { let processing_time_ms = processing_time.as_millis() as f64; // 更新性能指标（如果启用） - if self.config.enable_metrics { - self.metrics_manager - .update_metrics(event_count as u64, processing_time_ms) - .await; - } - + self.metrics_manager.update_metrics(event_count as u64, processing_time_ms).await; // 记录慢处理操作 self.metrics_manager.log_slow_processing(processing_time_ms, event_count); } EventPretty::BlockMeta(block_meta_pretty) => { + let start_time = std::time::Instant::now(); + self.metrics_manager.add_process_count().await; let block_time_ms = block_meta_pretty .block_time .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) @@ -96,6 +115,13 @@ impl EventProcessor { block_time_ms, ); callback(block_meta_event); + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + // 更新性能指标（如果启用） + self.metrics_manager.update_metrics(1, processing_time_ms).await; + // 记录慢处理操作 + self.metrics_manager.log_slow_processing(processing_time_ms, 1); } } @@ -114,7 +140,28 @@ impl EventProcessor { F: Fn(Vec>) + Send + Sync + 'static, { match event_pretty { + EventPretty::Account(account_pretty) => { + self.metrics_manager.add_process_count().await; + let start_time = std::time::Instant::now(); + let program_received_time_ms = chrono::Utc::now().timestamp_millis(); + let account_event = AccountEventParser::parse_account_event( + protocols.clone(), + account_pretty, + program_received_time_ms, + ); + if let Some(event) = account_event { + (batch_processor.callback)(vec![event]); + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + // 实际调用性能指标更新 + self.metrics_manager.update_metrics(1, processing_time_ms).await; + // 记录慢处理操作 + self.metrics_manager.log_slow_processing(processing_time_ms, 1); + } + } EventPretty::Transaction(transaction_pretty) => { + self.metrics_manager.add_process_count().await; let start_time = std::time::Instant::now(); let program_received_time_ms = chrono::Utc::now().timestamp_millis(); let slot = transaction_pretty.slot; @@ -142,8 +189,8 @@ impl EventProcessor { Ok(events) => { let event_count = events.len(); if !events.is_empty() { - log::info!("Parsed {} events", event_count); - log::info!("Adding {} events to batch processor", event_count); + log::debug!("Parsed {} events", event_count); + log::debug!("Adding {} events to batch processor", event_count); for event in events { if self.config.batch.enabled { batch_processor.add_event(event); @@ -165,7 +212,7 @@ impl EventProcessor { // 添加调试信息 if total_events > 0 { - log::info!( + log::debug!( "Total events parsed: {} for transaction {}", total_events, signature @@ -183,6 +230,7 @@ impl EventProcessor { self.metrics_manager.log_slow_processing(processing_time_ms, total_events); } EventPretty::BlockMeta(block_meta_pretty) => { + let start_time = std::time::Instant::now(); let block_time_ms = block_meta_pretty .block_time .map(|ts| ts.seconds * 1000 + ts.nanos as i64 / 1_000_000) @@ -193,6 +241,13 @@ impl EventProcessor { block_time_ms, ); (batch_processor.callback)(vec![block_meta_event]); + // 更新性能指标 + let processing_time = start_time.elapsed(); + let processing_time_ms = processing_time.as_millis() as f64; + // 更新性能指标（如果启用） + self.metrics_manager.update_metrics(1, processing_time_ms).await; + // 记录慢处理操作 + self.metrics_manager.log_slow_processing(processing_time_ms, 1); } } diff --git a/src/streaming/grpc/stream_handler.rs b/src/streaming/grpc/stream_handler.rs index 222995d..362d963 100644 --- a/src/streaming/grpc/stream_handler.rs +++ b/src/streaming/grpc/stream_handler.rs @@ -1,6 +1,5 @@ use chrono::Local; use futures::{channel::mpsc, sink::Sink, SinkExt}; -use log::info; use yellowstone_grpc_proto::geyser::{ subscribe_update::UpdateOneof, SubscribeRequest, SubscribeRequestPing, SubscribeUpdate, }; @@ -8,6 +7,7 @@ use yellowstone_grpc_proto::geyser::{ use super::types::{BlockMetaPretty, EventPretty, TransactionPretty}; use crate::common::AnyResult; use crate::streaming::common::BackpressureStrategy; +use crate::streaming::grpc::AccountPretty; /// 流消息处理器 pub struct StreamHandler; @@ -22,9 +22,19 @@ impl StreamHandler { ) -> AnyResult<()> { let created_at = msg.created_at; match msg.update_oneof { + Some(UpdateOneof::Account(account)) => { + let account_pretty = AccountPretty::from(account); + log::debug!("Received account: {:?}", account_pretty); + Self::handle_backpressure( + tx, + EventPretty::Account(account_pretty), + backpressure_strategy, + ) + .await?; + } Some(UpdateOneof::BlockMeta(sut)) => { let block_meta_pretty = BlockMetaPretty::from((sut, created_at)); - log::info!("Received block meta: {:?}", block_meta_pretty); + log::debug!("Received block meta: {:?}", block_meta_pretty); Self::handle_backpressure( tx, EventPretty::BlockMeta(block_meta_pretty), @@ -34,7 +44,7 @@ impl StreamHandler { } Some(UpdateOneof::Transaction(sut)) => { let transaction_pretty = TransactionPretty::from((sut, created_at)); - log::info!( + log::debug!( "Received transaction: {} at slot {}", transaction_pretty.signature, transaction_pretty.slot @@ -55,10 +65,10 @@ impl StreamHandler { ..Default::default() }) .await?; - info!("service is ping: {}", Local::now()); + log::debug!("service is ping: {}", Local::now()); } Some(UpdateOneof::Pong(_)) => { - info!("service is pong: {}", Local::now()); + log::debug!("service is pong: {}", Local::now()); } _ => { log::debug!("Received other message type"); diff --git a/src/streaming/grpc/subscription.rs b/src/streaming/grpc/subscription.rs index 2969a1c..9315ab2 100644 --- a/src/streaming/grpc/subscription.rs +++ b/src/streaming/grpc/subscription.rs @@ -4,10 +4,11 @@ use std::{collections::HashMap, time::Duration}; use tonic::{transport::channel::ClientTlsConfig, Status}; use yellowstone_grpc_client::{GeyserGrpcClient, Interceptor}; use yellowstone_grpc_proto::geyser::{ - CommitmentLevel, SubscribeRequest, SubscribeRequestFilterBlocksMeta, - SubscribeRequestFilterTransactions, SubscribeUpdate, + CommitmentLevel, SubscribeRequest, SubscribeRequestFilterAccounts, + SubscribeRequestFilterBlocksMeta, SubscribeRequestFilterTransactions, SubscribeUpdate, }; +use super::types::AccountsFilterMap; use super::types::TransactionsFilterMap; use crate::common::AnyResult; use crate::streaming::common::StreamClientConfig as ClientConfig; @@ -41,12 +42,14 @@ impl SubscriptionManager { pub async fn subscribe_with_request( &self, transactions: TransactionsFilterMap, + accounts: Option, commitment: Option, ) -> AnyResult<( impl Sink, impl Stream>, )> { let subscribe_request = SubscribeRequest { + accounts: accounts.unwrap_or_default(), transactions, blocks_meta: hashmap! { "".to_owned() => SubscribeRequestFilterBlocksMeta {} }, commitment: if let Some(commitment) = commitment { @@ -56,12 +59,33 @@ impl SubscriptionManager { }, ..Default::default() }; - let mut client = self.connect().await?; let (sink, stream) = client.subscribe_with_request(Some(subscribe_request)).await?; Ok((sink, stream)) } + /// 创建账户订阅请求并返回流 + pub fn subscribe_with_account_request( + &self, + account: Vec, + owner: Vec, + ) -> Option { + if account.len() == 0 && owner.len() == 0 { + return None; + } + let mut accounts = HashMap::new(); + accounts.insert( + "".to_owned(), + SubscribeRequestFilterAccounts { + account: account, + owner: owner, + filters: vec![], + nonempty_txn_signature: None, + }, + ); + Some(accounts) + } + /// 生成订阅请求过滤器 pub fn get_subscribe_request_filter( &self, @@ -84,21 +108,6 @@ impl SubscriptionManager { transactions } - /// 验证订阅参数 - pub fn validate_subscription_params( - &self, - account_include: &[String], - account_exclude: &[String], - account_required: &[String], - ) -> AnyResult<()> { - if account_include.is_empty() && account_exclude.is_empty() && account_required.is_empty() { - return Err(anyhow::anyhow!( - "account_include or account_exclude or account_required cannot be empty" - )); - } - Ok(()) - } - /// 获取配置 pub fn get_config(&self) -> &ClientConfig { &self.config diff --git a/src/streaming/grpc/types.rs b/src/streaming/grpc/types.rs index cd555d7..8d80a7c 100644 --- a/src/streaming/grpc/types.rs +++ b/src/streaming/grpc/types.rs @@ -3,17 +3,47 @@ use solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionE use std::{collections::HashMap, fmt}; use yellowstone_grpc_proto::{ geyser::{ - SubscribeRequestFilterTransactions, SubscribeUpdateBlockMeta, SubscribeUpdateTransaction, + SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions, SubscribeUpdateAccount, + SubscribeUpdateBlockMeta, SubscribeUpdateTransaction, }, prost_types::Timestamp, }; pub type TransactionsFilterMap = HashMap; +pub type AccountsFilterMap = HashMap; #[derive(Clone)] pub enum EventPretty { BlockMeta(BlockMetaPretty), Transaction(TransactionPretty), + Account(AccountPretty), +} + +#[derive(Clone)] +pub struct AccountPretty { + pub slot: u64, + pub signature: String, + pub pubkey: String, + pub executable: bool, + pub lamports: u64, + pub owner: String, + pub rent_epoch: u64, + pub data: Vec, +} + +impl fmt::Debug for AccountPretty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AccountPretty") + .field("slot", &self.slot) + .field("signature", &self.signature) + .field("pubkey", &self.pubkey) + .field("executable", &self.executable) + .field("lamports", &self.lamports) + .field("owner", &self.owner) + .field("rent_epoch", &self.rent_epoch) + .field("data", &self.data) + .finish() + } } #[derive(Clone)] @@ -62,6 +92,22 @@ impl fmt::Debug for TransactionPretty { } } +impl From for AccountPretty { + fn from(account: SubscribeUpdateAccount) -> Self { + let account_info = account.account.unwrap(); + Self { + slot: account.slot, + signature: bs58::encode(&account_info.txn_signature.unwrap_or_default()).into_string(), + pubkey: bs58::encode(&account_info.pubkey).into_string(), + executable: account_info.executable, + lamports: account_info.lamports, + owner: bs58::encode(&account_info.owner).into_string(), + rent_epoch: account_info.rent_epoch, + data: account_info.data, + } + } +} + impl From<(SubscribeUpdateBlockMeta, Option)> for BlockMetaPretty { fn from( (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): ( diff --git a/src/streaming/yellowstone_grpc.rs b/src/streaming/yellowstone_grpc.rs index d4261ce..4e09fc0 100644 --- a/src/streaming/yellowstone_grpc.rs +++ b/src/streaming/yellowstone_grpc.rs @@ -12,6 +12,19 @@ use crate::streaming::common::{ use crate::streaming::event_parser::{Protocol, UnifiedEvent}; use crate::streaming::grpc::{EventPretty, EventProcessor, StreamHandler, SubscriptionManager}; +/// 交易过滤器 +pub struct TransactionFilter { + pub account_include: Vec, + pub account_exclude: Vec, + pub account_required: Vec, +} + +/// 账户过滤器 +pub struct AccountFilter { + pub account: Vec, + pub owner: Vec, +} + #[derive(Clone)] pub struct YellowstoneGrpc { pub endpoint: String, @@ -103,9 +116,8 @@ impl YellowstoneGrpc { &self, protocols: Vec, bot_wallet: Option, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, + transaction_filter: TransactionFilter, + account_filter: AccountFilter, commitment: Option, callback: F, ) -> AnyResult<()> @@ -117,22 +129,20 @@ impl YellowstoneGrpc { self.metrics_manager.start_auto_monitoring().await; } - // 验证订阅参数 - self.subscription_manager.validate_subscription_params( - &account_include, - &account_exclude, - &account_required, - )?; - let transactions = self.subscription_manager.get_subscribe_request_filter( - account_include, - account_exclude, - account_required, + transaction_filter.account_include, + transaction_filter.account_exclude, + transaction_filter.account_required, ); + let accounts = self + .subscription_manager + .subscribe_with_account_request(account_filter.account, account_filter.owner); // 订阅事件 - let (mut subscribe_tx, mut stream) = - self.subscription_manager.subscribe_with_request(transactions, commitment).await?; + let (mut subscribe_tx, mut stream) = self + .subscription_manager + .subscribe_with_request(transactions, accounts, commitment) + .await?; // 创建通道，使用配置中的通道大小 let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); @@ -190,9 +200,8 @@ impl YellowstoneGrpc { &self, protocols: Vec, bot_wallet: Option, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, + transaction_filter: TransactionFilter, + account_filter: AccountFilter, commitment: Option, callback: F, ) -> AnyResult<()> @@ -204,22 +213,20 @@ impl YellowstoneGrpc { self.metrics_manager.start_auto_monitoring().await; } - // 验证订阅参数 - self.subscription_manager.validate_subscription_params( - &account_include, - &account_exclude, - &account_required, - )?; - let transactions = self.subscription_manager.get_subscribe_request_filter( - account_include, - account_exclude, - account_required, + transaction_filter.account_include, + transaction_filter.account_exclude, + transaction_filter.account_required, ); + let accounts = self + .subscription_manager + .subscribe_with_account_request(account_filter.account, account_filter.owner); // Subscribe to events - let (mut subscribe_tx, mut stream) = - self.subscription_manager.subscribe_with_request(transactions, commitment).await?; + let (mut subscribe_tx, mut stream) = self + .subscription_manager + .subscribe_with_request(transactions, accounts, commitment) + .await?; // Create channel let (mut tx, mut rx) = mpsc::channel::(self.config.backpressure.channel_size); @@ -287,33 +294,6 @@ impl YellowstoneGrpc { tokio::signal::ctrl_c().await?; Ok(()) } - - /// 默认订阅方法 - 委托给即时处理模式 - #[deprecated(since = "0.1.12", note = "Use subscribe_events_immediate instead")] - pub async fn subscribe_events_v2( - &self, - protocols: Vec, - bot_wallet: Option, - account_include: Vec, - account_exclude: Vec, - account_required: Vec, - commitment: Option, - callback: F, - ) -> AnyResult<()> - where - F: Fn(Box) + Send + Sync + 'static, - { - self.subscribe_events_immediate( - protocols, - bot_wallet, - account_include, - account_exclude, - account_required, - commitment, - callback, - ) - .await - } } // 实现 Clone trait 以支持模块间共享 diff --git a/src/streaming/yellowstone_sub_system.rs b/src/streaming/yellowstone_sub_system.rs index 3347239..2bdbc97 100755 --- a/src/streaming/yellowstone_sub_system.rs +++ b/src/streaming/yellowstone_sub_system.rs @@ -47,7 +47,7 @@ impl YellowstoneGrpc { addrs, ); let (mut subscribe_tx, mut stream) = - self.subscription_manager.subscribe_with_request(transactions, None).await?; + self.subscription_manager.subscribe_with_request(transactions, None, None).await?; let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); let callback = Box::new(callback);