Merge commit '4902c34bfabf081c614065f788b9bd4e649731b6'

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "solana-streamer-sdk"
-version = "0.4.0"
+version = "0.4.1"
 edition = "2021"
 authors = ["William <byteblock6@gmail.com>", "sgxiang <sgxiang@gmail.com>", "wei <1415121722@qq.com>"]
 repository = "https://github.com/0xfnzero/solana-streamer"
@@ -68,6 +68,7 @@ crossbeam = "0.8.4"
 crossbeam-queue = "0.3.12"
 parking_lot = "0.12.1"
 wide = "0.7"
+spl-token = "8.0.0"
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports"] }

README.md

@@ -46,14 +46,14 @@ Add the dependency to your `Cargo.toml`:
 
 ```toml
 # Add to your Cargo.toml
-solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.0" }
+solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" }
 ```
 
 ### Use crates.io
 
 ```toml
 # Add to your Cargo.toml
-solana-streamer-sdk = "0.4.0"
+solana-streamer-sdk = "0.4.1"
 ```
 
 ## Configuration System

README_CN.md

@@ -45,14 +45,14 @@ git clone https://github.com/0xfnzero/solana-streamer
 
 ```toml
 # 添加到您的 Cargo.toml
-solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.0" }
+solana-streamer-sdk = { path = "./solana-streamer", version = "0.4.1" }
 ```
 
 ### 使用 crates.io
 
 ```toml
 # 添加到您的 Cargo.toml
-solana-streamer-sdk = "0.4.0"
+solana-streamer-sdk = "0.4.1"
 ```
 
 ## 配置系统

src/main.rs

@@ -3,6 +3,7 @@ use solana_streamer_sdk::{
     streaming::{
         event_parser::{
             common::{filter::EventTypeFilter, EventType},
+            core::account_event_parser::CommonAccountEvent,
             protocols::{
                 bonk::{
                     parser::BONK_PROGRAM_ID, BonkGlobalConfigAccountEvent, BonkMigrateToAmmEvent,
@@ -196,7 +197,7 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
         match_event!(event, {
             // -------------------------- block meta -----------------------
             BlockMetaEvent => |e: BlockMetaEvent| {
-                println!("BlockMetaEvent: {:?}", e.metadata.program_handle_time_consuming_us);
+                println!("BlockMetaEvent: {:?}", e.metadata.handle_us);
             },
             // -------------------------- bonk -----------------------
             BonkPoolCreateEvent => |e: BonkPoolCreateEvent| {
@@ -333,6 +334,9 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
             RaydiumCpmmPoolStateAccountEvent => |e: RaydiumCpmmPoolStateAccountEvent| {
                 println!("RaydiumCpmmPoolStateAccountEvent: {e:?}");
             },
+            CommonAccountEvent => |e: CommonAccountEvent| {
+                println!("CommonAccountEvent: {e:?}");
+            },
         });
     }
 }

src/streaming/common/config.rs

@@ -26,7 +26,7 @@ pub struct BackpressureConfig {
 
 impl Default for BackpressureConfig {
     fn default() -> Self {
-        Self { permits: 1, strategy: BackpressureStrategy::default() }
+        Self { permits: 3000, strategy: BackpressureStrategy::default() }
     }
 }
 

src/streaming/common/event_processor.rs

@@ -1,6 +1,5 @@
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
-use std::time::Instant;
 
 use crossbeam_queue::SegQueue;
 use solana_sdk::pubkey::Pubkey;
@@ -13,7 +12,7 @@ use crate::streaming::common::{
 use crate::streaming::event_parser::common::filter::EventTypeFilter;
 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::core::traits::get_high_perf_clock;
+
 use crate::streaming::event_parser::EventParser;
 use crate::streaming::event_parser::{
     core::traits::UnifiedEvent, protocols::mutil::parser::MutilEventParser, Protocol,
@@ -31,19 +30,14 @@ pub struct EventProcessor {
     pub(crate) event_type_filter: Option<EventTypeFilter>,
     pub(crate) callback: Option<Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>>,
     pub(crate) backpressure_config: BackpressureConfig,
-    /// High-performance lockfree queue for gRPC events
     pub(crate) grpc_queue: Arc<SegQueue<(EventPretty, Option<Pubkey>)>>,
-    /// High-performance lockfree queue for shred events  
     pub(crate) shred_queue: Arc<SegQueue<(TransactionWithSlot, Option<Pubkey>)>>,
-    /// Fast O(1) counter for Drop strategy (avoids expensive SegQueue::len())
     pub(crate) grpc_pending_count: Arc<AtomicUsize>,
     pub(crate) shred_pending_count: Arc<AtomicUsize>,
-    /// Processing thread control
     pub(crate) processing_shutdown: Arc<AtomicBool>,
 }
 
 impl EventProcessor {
-    /// Create a new high-performance event processor
     pub fn new(metrics_manager: MetricsManager, config: ClientConfig) -> Self {
         let backpressure_config = config.backpressure.clone();
         let grpc_queue = Arc::new(SegQueue::new());
@@ -78,21 +72,15 @@ impl EventProcessor {
         self.protocols = protocols;
         self.event_type_filter = event_type_filter;
 
-        // Check if Block processing thread should be started (before moving backpressure_config)
-        let should_start_block_processing = true;
-        // matches!(backpressure_config.strategy, BackpressureStrategy::Block);
-
         self.backpressure_config = backpressure_config;
         self.callback = callback;
-        // Use stored values to initialize parser_cache
         let protocols_ref = &self.protocols;
         let event_type_filter_ref = self.event_type_filter.as_ref();
         self.parser_cache.get_or_init(|| {
             Arc::new(MutilEventParser::new(protocols_ref.clone(), event_type_filter_ref.cloned()))
         });
 
-        // Start Block processing thread if using Block strategy
-        if should_start_block_processing {
+        if matches!(self.backpressure_config.strategy, BackpressureStrategy::Block) {
             self.start_block_processing_thread();
         }
     }
@@ -101,13 +89,12 @@ impl EventProcessor {
         self.parser_cache.get().unwrap().clone()
     }
 
-    /// Create adapter callback
     fn create_adapter_callback(&self) -> Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync> {
         let callback = self.callback.clone().unwrap();
         let metrics_manager = self.metrics_manager.clone();
 
         Arc::new(move |event: Box<dyn UnifiedEvent>| {
-            let processing_time_us = event.program_handle_time_consuming_us() as f64;
+            let processing_time_us = event.handle_us() as f64;
             callback(event);
             metrics_manager.update_metrics(MetricsEventType::Transaction, 1, processing_time_us);
         })
@@ -121,7 +108,6 @@ impl EventProcessor {
         self.apply_backpressure_control(event_pretty, bot_wallet).await
     }
 
-    /// Apply backpressure control strategy  
     async fn apply_backpressure_control(
         &self,
         event_pretty: EventPretty,
@@ -129,7 +115,6 @@ impl EventProcessor {
     ) -> AnyResult<()> {
         match self.backpressure_config.strategy {
             BackpressureStrategy::Block => {
-                // Block strategy: async wait if queue is full (backpressure control)
                 loop {
                     let current_pending = self.grpc_pending_count.load(Ordering::Relaxed);
                     if current_pending < self.backpressure_config.permits {
@@ -137,14 +122,12 @@ impl EventProcessor {
                         self.grpc_pending_count.fetch_add(1, Ordering::Relaxed);
                         break;
                     }
-                    // Async yield to avoid blocking gRPC data source
+
                     tokio::task::yield_now().await;
                 }
                 Ok(())
             }
             BackpressureStrategy::Drop => {
-                // Drop strategy: Use O(1) atomic counter instead of expensive O(n) len()
-                // If pending count >= permits, DROP the event immediately
                 let current_pending = self.grpc_pending_count.load(Ordering::Relaxed);
                 if current_pending >= self.backpressure_config.permits {
                     self.metrics_manager.increment_dropped_events();
@@ -188,7 +171,7 @@ impl EventProcessor {
                     self.event_type_filter.as_ref(),
                 );
                 if let Some(event) = account_event {
-                    let processing_time_us = event.program_handle_time_consuming_us() as f64;
+                    let processing_time_us = event.handle_us() as f64;
                     self.invoke_callback(event);
                     self.update_metrics(MetricsEventType::Account, 1, processing_time_us);
                 }
@@ -197,20 +180,20 @@ impl EventProcessor {
                 self.metrics_manager.add_tx_process_count();
                 let slot = transaction_pretty.slot;
                 let signature = transaction_pretty.signature;
-                let tx = transaction_pretty.tx;
                 let block_time = transaction_pretty.block_time;
-                let program_received_time_us = transaction_pretty.program_received_time_us;
+                let recv_us = transaction_pretty.recv_us;
                 let transaction_index = transaction_pretty.transaction_index;
-                // Use cache to get parser
+                let grpc_tx = transaction_pretty.grpc_tx;
+
                 let parser = self.get_parser();
                 let adapter_callback = self.create_adapter_callback();
                 parser
-                    .parse_transaction_owned(
-                        tx,
+                    .parse_grpc_transaction_owned(
+                        grpc_tx,
                         signature,
                         Some(slot),
                         block_time,
-                        program_received_time_us,
+                        recv_us,
                         bot_wallet,
                         transaction_index,
                         adapter_callback,
@@ -225,11 +208,11 @@ impl EventProcessor {
                     .unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
                 let block_meta_event = CommonEventParser::generate_block_meta_event(
                     block_meta_pretty.slot,
-                    &block_meta_pretty.block_hash,
+                    block_meta_pretty.block_hash,
                     block_time_ms,
-                    block_meta_pretty.program_received_time_us,
+                    block_meta_pretty.recv_us,
                 );
-                let processing_time_us = block_meta_event.program_handle_time_consuming_us() as f64;
+                let processing_time_us = block_meta_event.handle_us() as f64;
                 self.invoke_callback(block_meta_event);
                 self.update_metrics(MetricsEventType::BlockMeta, 1, processing_time_us);
             }
@@ -244,7 +227,6 @@ impl EventProcessor {
         }
     }
 
-    /// Process a single transaction immediately
     pub async fn process_shred_transaction_immediate(
         &self,
         transaction_with_slot: TransactionWithSlot,
@@ -253,17 +235,14 @@ impl EventProcessor {
         self.process_shred_transaction(transaction_with_slot, bot_wallet).await
     }
 
-    /// Process shred transaction with backpressure control and performance monitoring
     pub async fn process_shred_transaction_with_metrics(
         &self,
         transaction_with_slot: TransactionWithSlot,
         bot_wallet: Option<Pubkey>,
     ) -> AnyResult<()> {
-        // Backpressure control logic
         self.apply_shred_backpressure_control(transaction_with_slot, bot_wallet).await
     }
 
-    /// Apply shred backpressure control strategy
     async fn apply_shred_backpressure_control(
         &self,
         transaction_with_slot: TransactionWithSlot,
@@ -271,7 +250,6 @@ impl EventProcessor {
     ) -> AnyResult<()> {
         match self.backpressure_config.strategy {
             BackpressureStrategy::Block => {
-                // Block strategy: async wait if queue is full (backpressure control)
                 loop {
                     let current_pending = self.shred_pending_count.load(Ordering::Relaxed);
                     if current_pending < self.backpressure_config.permits {
@@ -279,13 +257,12 @@ impl EventProcessor {
                         self.shred_pending_count.fetch_add(1, Ordering::Relaxed);
                         break;
                     }
-                    // Async yield to avoid blocking shred data source
+
                     tokio::task::yield_now().await;
                 }
                 Ok(())
             }
             BackpressureStrategy::Drop => {
-                // Drop strategy: Use O(1) atomic counter instead of expensive O(n) len()
                 let current_pending = self.shred_pending_count.load(Ordering::Relaxed);
                 if current_pending >= self.backpressure_config.permits {
                     self.metrics_manager.increment_dropped_events();
@@ -325,8 +302,8 @@ impl EventProcessor {
 
         let slot = transaction_with_slot.slot;
         let signature = tx.signatures[0];
-        let program_received_time_us = transaction_with_slot.program_received_time_us;
-        // Use cache to get parser
+        let recv_us = transaction_with_slot.recv_us;
+
         let parser = self.get_parser();
         let adapter_callback = self.create_adapter_callback();
         parser
@@ -335,7 +312,7 @@ impl EventProcessor {
                 signature,
                 Some(slot),
                 None,
-                program_received_time_us,
+                recv_us,
                 bot_wallet,
                 None,
                 &[],
@@ -350,9 +327,7 @@ impl EventProcessor {
         self.metrics_manager.update_metrics(ty, count, time_us);
     }
 
-    /// Start dedicated processing threads for all strategies
     fn start_block_processing_thread(&self) {
-        // Reset shutdown flag
         self.processing_shutdown.store(false, Ordering::Relaxed);
 
         let grpc_queue = Arc::clone(&self.grpc_queue);
@@ -363,36 +338,44 @@ impl EventProcessor {
         let shutdown_flag_clone = Arc::clone(&self.processing_shutdown);
         let processor = self.clone();
         let processor_clone = self.clone();
-        // 1. 专用线程 + 2. Busy-wait + 4. 无锁处理
+        // Dedicated thread with busy-wait and lock-free processing
         std::thread::spawn(move || {
-            // 创建blocking runtime for async processing
-            let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+            let worker_threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); // 如果获取失败则回退到4个线程
+
+            let rt = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(worker_threads)
+                .enable_all()
+                .build()
+                .unwrap();
+
             while !shutdown_flag.load(Ordering::Relaxed) {
                 if let Some((event_pretty, bot_wallet)) = grpc_queue.pop() {
-                    // Decrement pending counter when consuming from queue
                     grpc_pending_count.fetch_sub(1, Ordering::Relaxed);
-                    // Process event in blocking runtime
                     if let Err(e) = rt.block_on(
                         processor.process_grpc_event_transaction(event_pretty, bot_wallet),
                     ) {
                         println!("Error processing gRPC event: {}", e);
                     }
                 } else {
-                    // 2. 优化忙等待: 使用轻量级休眠减少CPU占用
+                    // Yield to reduce CPU usage in busy wait
                     std::thread::yield_now();
                 }
             }
         });
 
-        // Shred处理也使用相同的低延迟优化
+        // Shred processing with same low-latency optimization
         std::thread::spawn(move || {
-            let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+            let worker_threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); // 如果获取失败则回退到4个线程
+
+            let rt = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(worker_threads)
+                .enable_all()
+                .build()
+                .unwrap();
 
             while !shutdown_flag_clone.load(Ordering::Relaxed) {
                 if let Some((transaction_with_slot, bot_wallet)) = shred_queue.pop() {
-                    // Decrement pending counter when consuming from queue
                     shred_pending_count.fetch_sub(1, Ordering::Relaxed);
-                    // Process transaction in blocking runtime
                     if let Err(e) = rt.block_on(
                         processor_clone
                             .process_shred_transaction(transaction_with_slot, bot_wallet),
@@ -400,20 +383,18 @@ impl EventProcessor {
                         log::error!("Error processing shred transaction: {}", e);
                     }
                 } else {
-                    // 优化忙等待: 使用轻量级休眠减少CPU占用
+                    // Yield to reduce CPU usage in busy wait
                     std::thread::yield_now();
                 }
             }
         });
     }
 
-    /// Stop processing threads
     pub fn stop_processing(&self) {
         self.processing_shutdown.store(true, Ordering::Relaxed);
     }
 }
 
-// Implement Clone trait to support sharing between modules
 impl Clone for EventProcessor {
     fn clone(&self) -> Self {
         Self {

src/streaming/common/metrics.rs

@@ -39,7 +39,8 @@ struct AtomicEventMetrics {
     events_processed: AtomicU64,
     events_in_window: AtomicU64,
     window_start_nanos: AtomicU64,
-    events_per_second_bits: AtomicU64, // Bit representation of f64
+    // Processing time statistics per event type
+    processing_stats: AtomicProcessingTimeStats,
 }
 
 impl AtomicEventMetrics {
@@ -49,7 +50,7 @@ impl AtomicEventMetrics {
             events_processed: AtomicU64::new(0),
             events_in_window: AtomicU64::new(0),
             window_start_nanos: AtomicU64::new(now_nanos),
-            events_per_second_bits: AtomicU64::new(0),
+            processing_stats: AtomicProcessingTimeStats::new(),
         }
     }
 
@@ -76,18 +77,6 @@ impl AtomicEventMetrics {
         )
     }
 
-    /// Atomically update events per second
-    #[inline]
-    fn update_events_per_second(&self, eps: f64) {
-        self.events_per_second_bits.store(eps.to_bits(), Ordering::Relaxed);
-    }
-
-    /// Get events per second
-    #[inline]
-    fn get_events_per_second(&self) -> f64 {
-        f64::from_bits(self.events_per_second_bits.load(Ordering::Relaxed))
-    }
-
     /// Reset window count
     #[inline]
     fn reset_window(&self, new_start_nanos: u64) {
@@ -99,6 +88,18 @@ impl AtomicEventMetrics {
     fn get_window_start(&self) -> u64 {
         self.window_start_nanos.load(Ordering::Relaxed)
     }
+
+    /// Get processing time statistics for this event type
+    #[inline]
+    fn get_processing_stats(&self) -> ProcessingTimeStats {
+        self.processing_stats.get_stats()
+    }
+
+    /// Update processing time statistics for this event type
+    #[inline]
+    fn update_processing_stats(&self, time_us: f64, event_count: u64) {
+        self.processing_stats.update(time_us, event_count);
+    }
 }
 
 /// High-performance atomic processing time statistics
@@ -139,7 +140,7 @@ impl AtomicProcessingTimeStats {
         // Update minimum value, check time difference and reset if over 10 seconds
         let mut current_min = self.min_time_bits.load(Ordering::Relaxed);
         let min_timestamp = self.min_time_timestamp_nanos.load(Ordering::Relaxed);
-        
+
         // Check if min value timestamp exceeds 10 seconds (10_000_000_000 nanoseconds)
         let min_time_diff_nanos = now_nanos.saturating_sub(min_timestamp);
         if min_time_diff_nanos > 10_000_000_000 {
@@ -148,7 +149,7 @@ impl AtomicProcessingTimeStats {
             self.min_time_timestamp_nanos.store(now_nanos, Ordering::Relaxed);
             current_min = f64::INFINITY.to_bits();
         }
-        
+
         // If current time is less than min value, update min value and timestamp
         while time_bits < current_min {
             match self.min_time_bits.compare_exchange_weak(
@@ -236,7 +237,7 @@ pub struct ProcessingTimeStats {
 pub struct EventMetricsSnapshot {
     pub process_count: u64,
     pub events_processed: u64,
-    pub events_per_second: f64,
+    pub processing_stats: ProcessingTimeStats,
 }
 
 /// Compatibility structure - complete performance metrics
@@ -253,9 +254,12 @@ pub struct PerformanceMetrics {
 impl PerformanceMetrics {
     /// Create default performance metrics (compatibility method)
     pub fn new() -> Self {
-        let default_metrics =
-            EventMetricsSnapshot { process_count: 0, events_processed: 0, events_per_second: 0.0 };
         let default_stats = ProcessingTimeStats { min_us: 0.0, max_us: 0.0, avg_us: 0.0 };
+        let default_metrics = EventMetricsSnapshot {
+            process_count: 0,
+            events_processed: 0,
+            processing_stats: default_stats.clone(),
+        };
 
         Self {
             uptime: std::time::Duration::ZERO,
@@ -311,9 +315,9 @@ impl HighPerformanceMetrics {
     pub fn get_event_metrics(&self, event_type: EventType) -> EventMetricsSnapshot {
         let index = event_type.as_index();
         let (process_count, events_processed, _) = self.event_metrics[index].get_counts();
-        let events_per_second = self.calculate_real_time_eps(event_type);
+        let processing_stats = self.event_metrics[index].get_processing_stats();
 
-        EventMetricsSnapshot { process_count, events_processed, events_per_second }
+        EventMetricsSnapshot { process_count, events_processed, processing_stats }
     }
 
     /// 获取处理时间统计
@@ -328,41 +332,6 @@ impl HighPerformanceMetrics {
         self.dropped_events_count.load(Ordering::Relaxed)
     }
 
-    /// 计算实时每秒事件数（非阻塞）
-    fn calculate_real_time_eps(&self, event_type: EventType) -> f64 {
-        let now_nanos =
-            std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()
-                as u64;
-
-        let index = event_type.as_index();
-        let event_metric = &self.event_metrics[index];
-
-        let window_start = event_metric.get_window_start();
-        let current_window_duration_secs =
-            (now_nanos.saturating_sub(window_start)) as f64 / 1_000_000_000.0;
-        let events_in_window = event_metric.events_in_window.load(Ordering::Relaxed);
-
-        // 优先级1: 当前窗口实时数据（≥2秒且有事件）
-        if current_window_duration_secs >= 2.0 && events_in_window > 0 {
-            return events_in_window as f64 / current_window_duration_secs;
-        }
-
-        // 优先级2: 上一个窗口的结果
-        let stored_eps = event_metric.get_events_per_second();
-        if stored_eps > 0.0 {
-            return stored_eps;
-        }
-
-        // 优先级3: 总体平均值（≥3秒运行时间）
-        let total_duration_secs = self.get_uptime_seconds();
-        let total_events = event_metric.events_processed.load(Ordering::Relaxed);
-        if total_duration_secs >= 3.0 && total_events > 0 {
-            return total_events as f64 / total_duration_secs;
-        }
-
-        0.0
-    }
-
     /// 更新窗口指标（后台任务调用）
     fn update_window_metrics(&self, event_type: EventType, window_duration_nanos: u64) {
         let now_nanos =
@@ -374,14 +343,6 @@ impl HighPerformanceMetrics {
 
         let window_start = event_metric.get_window_start();
         if now_nanos.saturating_sub(window_start) >= window_duration_nanos {
-            let events_in_window = event_metric.events_in_window.load(Ordering::Relaxed);
-            let window_duration_secs = window_duration_nanos as f64 / 1_000_000_000.0;
-
-            if window_duration_secs > 0.001 && events_in_window > 0 {
-                let eps = events_in_window as f64 / window_duration_secs;
-                event_metric.update_events_per_second(eps);
-            }
-
             event_metric.reset_window(now_nanos);
         }
     }
@@ -455,10 +416,15 @@ impl MetricsManager {
             return;
         }
 
-        // 原子更新事件计数
-        self.metrics.event_metrics[event_type.as_index()].add_events_processed(count);
+        let index = event_type.as_index();
 
-        // 原子更新处理时间统计
+        // 原子更新事件计数
+        self.metrics.event_metrics[index].add_events_processed(count);
+
+        // 原子更新该事件类型的处理时间统计
+        self.metrics.event_metrics[index].update_processing_stats(processing_time_us, count);
+
+        // 保持全局处理时间统计的兼容性
         self.metrics.processing_stats.update(processing_time_us, count);
     }
 
@@ -506,35 +472,25 @@ impl MetricsManager {
             println!("\n⚠️  Dropped Events: {}", dropped_count);
         }
 
-        // 打印事件指标表格
-        println!("┌─────────────┬──────────────┬──────────────────┬─────────────────┐");
-        println!("│ Event Type  │ Process Count│ Events Processed │ Events/Second   │");
-        println!("├─────────────┼──────────────┼──────────────────┼─────────────────┤");
+        // 打印事件指标表格（包含处理时间统计）
+        println!("┌─────────────┬──────────────┬──────────────────┬─────────────┬─────────────┬─────────────┐");
+        println!("│ Event Type  │ Process Count│ Events Processed │ Avg Time(μs)│ Min 10s(μs) │ Max 10s(μs) │");
+        println!("├─────────────┼──────────────┼──────────────────┼─────────────┼─────────────┼─────────────┤");
 
         for event_type in [EventType::Transaction, EventType::Account, EventType::BlockMeta] {
             let metrics = self.get_event_metrics(event_type);
             println!(
-                "│ {:11} │ {:12} │ {:16} │ {:13.2}   │",
+                "│ {:11} │ {:12} │ {:16} │ {:9.2}   │ {:9.2}   │ {:9.2}   │",
                 event_type.name(),
                 metrics.process_count,
                 metrics.events_processed,
-                metrics.events_per_second
+                metrics.processing_stats.avg_us,
+                metrics.processing_stats.min_us,
+                metrics.processing_stats.max_us
             );
         }
 
-        println!("└─────────────┴──────────────┴──────────────────┴─────────────────┘");
-
-        // 打印处理时间统计表格
-        let stats = self.get_processing_stats();
-        println!("\n⏱️  Processing Time Statistics");
-        println!("┌───────────────────────┬─────────────┐");
-        println!("│ Metric                │ Value (us)  │");
-        println!("├───────────────────────┼─────────────┤");
-        println!("│ Average               │ {:9.2}   │", stats.avg_us);
-        println!("│ Minimum within 10s    │ {:9.2}   │", stats.min_us);
-        println!("│ Maximum within 10s    │ {:9.2}   │", stats.max_us);
-        println!("└───────────────────────┴─────────────┘");
-
+        println!("└─────────────┴──────────────┴──────────────────┴─────────────┴─────────────┴─────────────┘");
         println!();
     }
 

src/streaming/event_parser/common/mod.rs

@@ -12,7 +12,7 @@ macro_rules! impl_unified_event {
                 self.metadata.event_type.clone()
             }
 
-            fn signature(&self) -> &str {
+            fn signature(&self) -> &solana_sdk::signature::Signature {
                 &self.metadata.signature
             }
 
@@ -20,16 +20,16 @@ macro_rules! impl_unified_event {
                 self.metadata.slot
             }
 
-            fn program_received_time_us(&self) -> i64 {
-                self.metadata.program_received_time_us
+            fn recv_us(&self) -> i64 {
+                self.metadata.recv_us
             }
 
-            fn program_handle_time_consuming_us(&self) -> i64 {
-                self.metadata.program_handle_time_consuming_us
+            fn handle_us(&self) -> i64 {
+                self.metadata.handle_us
             }
 
-            fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64) {
-                self.metadata.program_handle_time_consuming_us = program_handle_time_consuming_us;
+            fn set_handle_us(&mut self, handle_us: i64) {
+                self.metadata.handle_us = handle_us;
             }
 
             fn as_any(&self) -> &dyn std::any::Any {
@@ -60,12 +60,12 @@ macro_rules! impl_unified_event {
                 self.metadata.swap_data.is_some()
             }
 
-            fn instruction_outer_index(&self) -> i64 {
-                self.metadata.instruction_outer_index
+            fn outer_index(&self) -> i64 {
+                self.metadata.outer_index
             }
 
-            fn instruction_inner_index(&self) -> Option<i64> {
-                self.metadata.instruction_inner_index
+            fn inner_index(&self) -> Option<i64> {
+                self.metadata.inner_index
             }
             fn transaction_index(&self) -> Option<u64> {
                 self.metadata.transaction_index

src/streaming/event_parser/common/types.rs

@@ -1,7 +1,7 @@
 use borsh::{BorshDeserialize, BorshSerialize};
 use crossbeam_queue::ArrayQueue;
 use serde::{Deserialize, Serialize};
-use solana_sdk::pubkey::Pubkey;
+use solana_sdk::{pubkey::Pubkey, signature::Signature};
 use std::{borrow::Cow, fmt, str::FromStr, sync::Arc};
 
 use crate::{
@@ -141,6 +141,8 @@ pub enum EventType {
     AccountRaydiumCpmmAmmConfig,
     AccountRaydiumCpmmPoolState,
 
+    AccountCommon,
+
     // Common events
     BlockMeta,
     Unknown,
@@ -225,6 +227,7 @@ impl fmt::Display for EventType {
             }
             EventType::AccountRaydiumCpmmAmmConfig => write!(f, "AccountRaydiumCpmmAmmConfig"),
             EventType::AccountRaydiumCpmmPoolState => write!(f, "AccountRaydiumCpmmPoolState"),
+            EventType::AccountCommon => write!(f, "AccountCommon"),
             EventType::BlockMeta => write!(f, "BlockMeta"),
             EventType::Unknown => write!(f, "Unknown"),
         }
@@ -282,42 +285,40 @@ pub struct SwapData {
     pub to_mint: Pubkey,
     pub from_amount: u64,
     pub to_amount: u64,
-    pub description: Option<String>,
+    pub description: Option<Cow<'static, str>>,
 }
 
 /// Event metadata
-#[derive(
-    Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, BorshSerialize, BorshDeserialize,
-)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct EventMetadata {
-    pub signature: Cow<'static, str>,
+    pub signature: Signature,
     pub slot: u64,
     pub transaction_index: Option<u64>, // 新增：交易在slot中的索引
     pub block_time: i64,
     pub block_time_ms: i64,
-    pub program_received_time_us: i64,
-    pub program_handle_time_consuming_us: i64,
+    pub recv_us: i64,
+    pub handle_us: i64,
     pub protocol: ProtocolType,
     pub event_type: EventType,
     pub program_id: Pubkey,
     pub swap_data: Option<SwapData>,
-    pub instruction_outer_index: i64,
-    pub instruction_inner_index: Option<i64>,
+    pub outer_index: i64,
+    pub inner_index: Option<i64>,
 }
 
 impl EventMetadata {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
-        signature: Cow<'static, str>,
+        signature: Signature,
         slot: u64,
         block_time: i64,
         block_time_ms: i64,
         protocol: ProtocolType,
         event_type: EventType,
         program_id: Pubkey,
-        instruction_outer_index: i64,
-        instruction_inner_index: Option<i64>,
-        program_received_time_us: i64,
+        outer_index: i64,
+        inner_index: Option<i64>,
+        recv_us: i64,
         transaction_index: Option<u64>,
     ) -> Self {
         Self {
@@ -325,14 +326,14 @@ impl EventMetadata {
             slot,
             block_time,
             block_time_ms,
-            program_received_time_us,
-            program_handle_time_consuming_us: 0,
+            recv_us,
+            handle_us: 0,
             protocol,
             event_type,
             program_id,
             swap_data: None,
-            instruction_outer_index,
-            instruction_inner_index,
+            outer_index,
+            inner_index,
             transaction_index,
         }
     }
@@ -413,7 +414,7 @@ pub fn parse_swap_data_from_next_instructions(
         },
         RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| {
             user = Some(e.payer);
-            swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".to_string());
+            swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into());
             user_from_token = Some(e.input_token_account);
             user_to_token   = Some(e.output_token_account);
             from_vault = Some(e.input_vault);
@@ -430,7 +431,7 @@ pub fn parse_swap_data_from_next_instructions(
         },
         RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| {
             user = Some(e.user_source_owner);
-            swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".to_string());
+            swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into());
             user_from_token = Some(e.user_source_token_account);
             user_to_token   = Some(e.user_destination_token_account);
             from_vault = Some(e.pool_pc_token_account);
@@ -453,12 +454,179 @@ pub fn parse_swap_data_from_next_instructions(
             break;
         }
         let data = &compiled.data;
-        
+
         // 使用 SIMD 验证数据格式
         if !SimdUtils::validate_data_format(data, 8) {
             continue;
         }
-        
+
+        let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize];
+        let (source, destination, amount) = match data[0] {
+            12 if compiled.accounts.len() >= 4 => {
+                let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
+                (get_pubkey(0), get_pubkey(2), amt)
+            }
+            3 if compiled.accounts.len() >= 3 => {
+                let amt = u64::from_le_bytes(data[1..9].try_into().unwrap());
+                (get_pubkey(0), get_pubkey(1), amt)
+            }
+            2 if compiled.accounts.len() >= 2 => {
+                let amt = u64::from_le_bytes(data[4..12].try_into().unwrap());
+                (get_pubkey(0), get_pubkey(1), amt)
+            }
+            _ => continue,
+        };
+
+        match (source, destination) {
+            (s, d) if s == user_to_token && d == to_vault => {
+                swap_data.from_mint = to_mint;
+                swap_data.from_amount = amount;
+            }
+            (s, d) if s == from_vault && d == user_from_token => {
+                swap_data.to_mint = from_mint;
+                swap_data.to_amount = amount;
+            }
+            (s, d) if s == user_from_token && d == from_vault => {
+                swap_data.from_mint = from_mint;
+                swap_data.from_amount = amount;
+            }
+            (s, d) if s == to_vault && d == user_to_token => {
+                swap_data.to_mint = to_mint;
+                swap_data.to_amount = amount;
+            }
+            (s, d) if s == user_from_token && d == to_vault => {
+                swap_data.from_mint = from_mint;
+                swap_data.from_amount = amount;
+            }
+            (s, d) if s == from_vault && d == user_to_token => {
+                swap_data.to_mint = to_mint;
+                swap_data.to_amount = amount;
+            }
+            _ => {}
+        }
+        if swap_data.from_mint != Pubkey::default() && swap_data.to_mint != Pubkey::default() {
+            break;
+        }
+        if swap_data.from_amount != 0 && swap_data.to_amount != 0 {
+            break;
+        }
+    }
+
+    if swap_data.from_mint != Pubkey::default()
+        || swap_data.to_mint != Pubkey::default()
+        || swap_data.from_amount != 0
+        || swap_data.to_amount != 0
+    {
+        Some(swap_data)
+    } else {
+        None
+    }
+}
+
+/// Parse token transfer data from next instructions
+/// TODO: - wait refactor
+pub fn parse_swap_data_from_next_grpc_instructions(
+    event: &dyn UnifiedEvent,
+    inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstructions,
+    current_index: i8,
+    accounts: &[Pubkey],
+) -> Option<SwapData> {
+    let mut swap_data = SwapData {
+        from_mint: Pubkey::default(),
+        to_mint: Pubkey::default(),
+        from_amount: 0,
+        to_amount: 0,
+        description: None,
+    };
+
+    // 先根据 event 取出关键信息
+    let mut user: Option<Pubkey> = None;
+    let mut from_mint: Option<Pubkey> = None;
+    let mut to_mint: Option<Pubkey> = None;
+    let mut user_from_token: Option<Pubkey> = None;
+    let mut user_to_token: Option<Pubkey> = None;
+    let mut from_vault: Option<Pubkey> = None;
+    let mut to_vault: Option<Pubkey> = None;
+
+    match_event!(&*event, {
+        BonkTradeEvent => |e: BonkTradeEvent| {
+            user = Some(e.payer);
+            from_mint = Some(e.base_token_mint);
+            to_mint = Some(e.quote_token_mint);
+            user_from_token = Some(e.user_base_token);
+            user_to_token = Some(e.user_quote_token);
+            from_vault = Some(e.base_vault);
+            to_vault = Some(e.quote_vault);
+        },
+        PumpFunTradeEvent => |e: PumpFunTradeEvent| {
+            swap_data.from_mint = if e.is_buy { *SOL_MINT } else { e.mint };
+            swap_data.to_mint   = if e.is_buy { e.mint } else { *SOL_MINT };
+        },
+        PumpSwapBuyEvent => |e: PumpSwapBuyEvent| {
+            swap_data.from_mint = e.quote_mint;
+            swap_data.to_mint   = e.base_mint;
+        },
+        PumpSwapSellEvent => |e: PumpSwapSellEvent| {
+            swap_data.from_mint = e.base_mint;
+            swap_data.to_mint   = e.quote_mint;
+        },
+        RaydiumCpmmSwapEvent => |e: RaydiumCpmmSwapEvent| {
+            user = Some(e.payer);
+            from_mint = Some(e.input_token_mint);
+            to_mint   = Some(e.output_token_mint);
+            user_from_token = Some(e.input_token_account);
+            user_to_token   = Some(e.output_token_account);
+            from_vault = Some(e.input_vault);
+            to_vault   = Some(e.output_vault);
+        },
+        RaydiumClmmSwapEvent => |e: RaydiumClmmSwapEvent| {
+            user = Some(e.payer);
+            swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumClmmSwapEvent".into());
+            user_from_token = Some(e.input_token_account);
+            user_to_token   = Some(e.output_token_account);
+            from_vault = Some(e.input_vault);
+            to_vault   = Some(e.output_vault);
+        },
+        RaydiumClmmSwapV2Event => |e: RaydiumClmmSwapV2Event| {
+            user = Some(e.payer);
+            from_mint = Some(e.input_vault_mint);
+            to_mint   = Some(e.output_vault_mint);
+            user_from_token = Some(e.input_token_account);
+            user_to_token   = Some(e.output_token_account);
+            from_vault = Some(e.input_vault);
+            to_vault   = Some(e.output_vault);
+        },
+        RaydiumAmmV4SwapEvent => |e: RaydiumAmmV4SwapEvent| {
+            user = Some(e.user_source_owner);
+            swap_data.description = Some("Unable to get from_mint and to_mint from RaydiumAmmV4SwapEvent".into());
+            user_from_token = Some(e.user_source_token_account);
+            user_to_token   = Some(e.user_destination_token_account);
+            from_vault = Some(e.pool_pc_token_account);
+            to_vault   = Some(e.pool_coin_token_account);
+        },
+    });
+
+    let user_to_token = user_to_token.unwrap_or_default();
+    let user_from_token = user_from_token.unwrap_or_default();
+    let to_vault = to_vault.unwrap_or_default();
+    let from_vault = from_vault.unwrap_or_default();
+    let to_mint = to_mint.unwrap_or_default();
+    let from_mint = from_mint.unwrap_or_default();
+
+    // 单次循环完成提取和判断
+    for instruction in inner_instruction.instructions.iter().skip((current_index + 1) as usize) {
+        let compiled = &instruction;
+        let program_id = accounts[compiled.program_id_index as usize];
+        if !SYSTEM_PROGRAMS.contains(&program_id) {
+            break;
+        }
+        let data = &compiled.data;
+
+        // 使用 SIMD 验证数据格式
+        if !SimdUtils::validate_data_format(data, 8) {
+            continue;
+        }
+
         let get_pubkey = |i: usize| accounts[compiled.accounts[i] as usize];
         let (source, destination, amount) = match data[0] {
             12 if compiled.accounts.len() >= 4 => {

src/streaming/event_parser/core/account_event_parser.rs

@@ -1,13 +1,16 @@
-use std::borrow::Cow;
 use std::collections::HashMap;
 use std::sync::OnceLock;
 
+use serde::{Deserialize, Serialize};
+use solana_sdk::program_pack::Pack;
 use solana_sdk::pubkey::Pubkey;
+use spl_token::state::Account;
 
+use crate::impl_unified_event;
 use crate::streaming::common::SimdUtils;
 use crate::streaming::event_parser::common::filter::EventTypeFilter;
 use crate::streaming::event_parser::common::{EventMetadata, EventType, ProtocolType};
-use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock};
+use crate::streaming::event_parser::core::traits::{elapsed_micros_since, 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;
@@ -27,6 +30,19 @@ pub struct AccountEventParseConfig {
     pub account_parser: AccountEventParserFn,
 }
 
+/// 通用账户事件
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CommonAccountEvent {
+    pub metadata: EventMetadata,
+    pub pubkey: Pubkey,
+    pub executable: bool,
+    pub lamports: u64,
+    pub owner: Pubkey,
+    pub rent_epoch: u64,
+    pub amount: Option<u64>,
+}
+impl_unified_event!(CommonAccountEvent,);
+
 /// 账户事件解析器
 pub type AccountEventParserFn =
     fn(account: &AccountPretty, metadata: EventMetadata) -> Option<Box<dyn UnifiedEvent>>;
@@ -34,6 +50,9 @@ pub type AccountEventParserFn =
 static PROTOCOL_CONFIGS_CACHE: OnceLock<HashMap<Protocol, Vec<AccountEventParseConfig>>> =
     OnceLock::new();
 
+// 通用账户解析配置的静态缓存
+static COMMON_CONFIG: OnceLock<AccountEventParseConfig> = OnceLock::new();
+
 pub struct AccountEventParser {}
 
 impl AccountEventParser {
@@ -149,21 +168,39 @@ impl AccountEventParser {
             map
         });
 
-        let mut configs = vec![];
-        let empty_vec = vec![];
+        let mut configs = Vec::new();
+        let empty_vec = Vec::new();
+
+        // 预估容量以减少重新分配
+        let estimated_capacity = protocols.len() * 3; // 大多数协议有2-3个配置
+        configs.reserve(estimated_capacity);
+
         for protocol in protocols {
             let protocol_configs = protocols_map.get(protocol).unwrap_or(&empty_vec);
-            let filtered_configs: Vec<AccountEventParseConfig> = protocol_configs
-                .iter()
-                .filter(|config| {
-                    event_type_filter
-                        .map(|filter| filter.include.contains(&config.event_type))
-                        .unwrap_or(true)
-                })
-                .cloned()
-                .collect();
-            configs.extend(filtered_configs);
+            // 如果没有过滤器，直接扩展所有配置
+            if event_type_filter.is_none() {
+                configs.extend(protocol_configs.iter().cloned());
+            } else {
+                // 有过滤器时才进行过滤
+                let filter = event_type_filter.unwrap();
+                configs.extend(
+                    protocol_configs
+                        .iter()
+                        .filter(|config| filter.include.contains(&config.event_type))
+                        .cloned(),
+                );
+            }
         }
+
+        let common_config = COMMON_CONFIG.get_or_init(|| AccountEventParseConfig {
+            program_id: Pubkey::default(),
+            protocol_type: ProtocolType::Common,
+            event_type: EventType::AccountCommon,
+            account_discriminator: &[],
+            account_parser: Self::parse_token_account_event,
+        });
+        configs.push(common_config.clone());
+
         configs
     }
 
@@ -174,30 +211,49 @@ impl AccountEventParser {
     ) -> Option<Box<dyn UnifiedEvent>> {
         let configs = Self::configs(protocols, event_type_filter);
         for config in configs {
-            if account.owner == config.program_id
-                && SimdUtils::fast_discriminator_match(&account.data, config.account_discriminator)
+            if config.program_id == Pubkey::default()
+                || (account.owner == config.program_id
+                    && SimdUtils::fast_discriminator_match(
+                        &account.data,
+                        config.account_discriminator,
+                    ))
             {
-                let signature_str = Cow::Owned(account.signature.to_string());
                 let event = (config.account_parser)(
                     &account,
                     EventMetadata {
                         slot: account.slot,
-                        signature: signature_str,
+                        signature: account.signature,
                         protocol: config.protocol_type,
                         event_type: config.event_type,
                         program_id: config.program_id,
-                        program_received_time_us: account.program_received_time_us,
+                        recv_us: account.recv_us,
                         ..Default::default()
                     },
                 );
                 if let Some(mut event) = event {
-                    event.set_program_handle_time_consuming_us(
-                        get_high_perf_clock().elapsed_micros_since(account.program_received_time_us),
-                    );
+                    event.set_handle_us(elapsed_micros_since(account.recv_us));
                     return Some(event);
                 }
             }
         }
         None
     }
+
+    pub fn parse_token_account_event(
+        account: &AccountPretty,
+        metadata: EventMetadata,
+    ) -> Option<Box<dyn UnifiedEvent>> {
+        let info = Account::unpack(&account.data);
+        let mut event = CommonAccountEvent {
+            metadata,
+            pubkey: account.pubkey,
+            executable: account.executable,
+            lamports: account.lamports,
+            owner: account.owner,
+            rent_epoch: account.rent_epoch,
+            amount: if let Ok(info) = info { Some(info.amount) } else { None },
+        };
+        event.set_handle_us(elapsed_micros_since(account.recv_us));
+        return Some(Box::new(event));
+    }
 }

src/streaming/event_parser/core/common_event_parser.rs

@@ -1,4 +1,4 @@
-use crate::streaming::event_parser::core::traits::{UnifiedEvent, get_high_perf_clock};
+use crate::streaming::event_parser::core::traits::{elapsed_micros_since, UnifiedEvent};
 use crate::streaming::event_parser::protocols::block::block_meta_event::BlockMetaEvent;
 
 pub struct CommonEventParser {}
@@ -6,19 +6,14 @@ pub struct CommonEventParser {}
 impl CommonEventParser {
     pub fn generate_block_meta_event(
         slot: u64,
-        block_hash: &str,
+        block_hash: String,
         block_time_ms: i64,
-        program_received_time_us: i64,
+        recv_us: i64,
     ) -> Box<dyn UnifiedEvent> {
-        let mut block_meta_event = BlockMetaEvent::new(
-            slot,
-            block_hash.to_string(),
-            block_time_ms,
-            program_received_time_us,
-        );
-        block_meta_event.set_program_handle_time_consuming_us(
-            get_high_perf_clock().elapsed_micros_since(program_received_time_us),
-        );
+        let mut block_meta_event =
+            BlockMetaEvent::new(slot, block_hash, block_time_ms, recv_us);
+        block_meta_event
+            .set_handle_us(elapsed_micros_since(recv_us));
         Box::new(block_meta_event)
     }
 }

src/streaming/event_parser/core/macros.rs

@@ -30,7 +30,7 @@ macro_rules! impl_event_parser_delegate {
                 signature: solana_sdk::signature::Signature,
                 slot: u64,
                 block_time: Option<prost_types::Timestamp>,
-                program_received_time_us: i64,
+                recv_us: i64,
                 outer_index: i64,
                 inner_index: Option<i64>,
                 transaction_index: Option<u64>,
@@ -41,7 +41,7 @@ macro_rules! impl_event_parser_delegate {
                     signature,
                     slot,
                     block_time,
-                    program_received_time_us,
+                    recv_us,
                     outer_index,
                     inner_index,
                     transaction_index,
@@ -49,6 +49,21 @@ macro_rules! impl_event_parser_delegate {
                 )
             }
 
+            fn parse_events_from_grpc_inner_instruction(
+                &self,
+                inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction,
+                signature: solana_sdk::signature::Signature,
+                slot: u64,
+                block_time: Option<prost_types::Timestamp>,
+                recv_us: i64,
+                outer_index: i64,
+                inner_index: Option<i64>,
+                transaction_index: Option<u64>,
+                config: &GenericEventParseConfig,
+            ) -> Vec<Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>> {
+                self.inner.parse_events_from_grpc_inner_instruction(inner_instruction, signature, slot, block_time, recv_us, outer_index, inner_index, transaction_index, config)
+            }
+
             fn parse_events_from_instruction(
                 &self,
                 instruction: &solana_sdk::instruction::CompiledInstruction,
@@ -56,7 +71,7 @@ macro_rules! impl_event_parser_delegate {
                 signature: solana_sdk::signature::Signature,
                 slot: u64,
                 block_time: Option<prost_types::Timestamp>,
-                program_received_time_us: i64,
+                recv_us: i64,
                 outer_index: i64,
                 inner_index: Option<i64>,
                 bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
@@ -74,7 +89,7 @@ macro_rules! impl_event_parser_delegate {
                     signature,
                     slot,
                     block_time,
-                    program_received_time_us,
+                    recv_us,
                     outer_index,
                     inner_index,
                     bot_wallet,
@@ -84,6 +99,28 @@ macro_rules! impl_event_parser_delegate {
                 )
             }
 
+            fn parse_events_from_grpc_instruction(
+                &self,
+                instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
+                accounts: &[solana_sdk::pubkey::Pubkey],
+                signature: solana_sdk::signature::Signature,
+                slot: u64,
+                block_time: Option<prost_types::Timestamp>,
+                recv_us: i64,
+                outer_index: i64,
+                inner_index: Option<i64>,
+                bot_wallet: Option<solana_sdk::pubkey::Pubkey>,
+                transaction_index: Option<u64>,
+                inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
+                callback: std::sync::Arc<
+                    dyn for<'a> Fn(&'a Box<dyn $crate::streaming::event_parser::core::traits::UnifiedEvent>)
+                        + Send
+                        + Sync,
+                >,
+            ) -> anyhow::Result<()> {
+                self.inner.parse_events_from_grpc_instruction(instruction, accounts, signature, slot, block_time, recv_us, outer_index, inner_index, bot_wallet, transaction_index, inner_instructions, callback)
+            }
+
             fn should_handle(&self, program_id: &solana_sdk::pubkey::Pubkey) -> bool {
                 self.inner.should_handle(program_id)
             }

src/streaming/event_parser/core/traits.rs

@@ -13,13 +13,16 @@ use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Instant;
+use yellowstone_grpc_proto::geyser::SubscribeUpdateTransactionInfo;
 
 use super::global_state::{
     add_bonk_dev_address, add_dev_address, is_bonk_dev_address, is_dev_address,
 };
 
 use crate::streaming::common::simd_utils::SimdUtils;
-use crate::streaming::event_parser::common::{parse_swap_data_from_next_instructions, SwapData};
+use crate::streaming::event_parser::common::{
+    parse_swap_data_from_next_grpc_instructions, parse_swap_data_from_next_instructions, SwapData,
+};
 use crate::streaming::event_parser::protocols::pumpswap::{PumpSwapBuyEvent, PumpSwapSellEvent};
 use crate::streaming::event_parser::{
     common::{EventMetadata, EventType, ProtocolType},
@@ -29,22 +32,53 @@ use crate::streaming::event_parser::{
     },
 };
 
-/// 高性能时钟管理器，减少系统调用开销
+/// 高性能时钟管理器，减少系统调用开销并最小化延迟
 #[derive(Debug)]
 pub struct HighPerformanceClock {
     /// 基准时间点（程序启动时的单调时钟时间）
     base_instant: Instant,
     /// 基准时间点对应的UTC时间戳（微秒）
     base_timestamp_us: i64,
+    /// 上次校准时间（用于检测是否需要重新校准）
+    last_calibration: Instant,
+    /// 校准间隔（秒）
+    calibration_interval_secs: u64,
 }
 
 impl HighPerformanceClock {
     /// 创建新的高性能时钟
     pub fn new() -> Self {
-        let base_instant = Instant::now();
-        let base_timestamp_us = chrono::Utc::now().timestamp_micros();
+        Self::new_with_calibration_interval(300) // 默认5分钟校准一次
+    }
 
-        Self { base_instant, base_timestamp_us }
+    /// 创建带自定义校准间隔的高性能时钟
+    pub fn new_with_calibration_interval(calibration_interval_secs: u64) -> Self {
+        // 通过多次采样来减少初始化误差
+        let mut best_offset = i64::MAX;
+        let mut best_instant = Instant::now();
+        let mut best_timestamp = chrono::Utc::now().timestamp_micros();
+
+        // 进行3次采样，选择延迟最小的
+        for _ in 0..3 {
+            let instant_before = Instant::now();
+            let timestamp = chrono::Utc::now().timestamp_micros();
+            let instant_after = Instant::now();
+
+            let sample_latency = instant_after.duration_since(instant_before).as_nanos() as i64;
+
+            if sample_latency < best_offset {
+                best_offset = sample_latency;
+                best_instant = instant_before;
+                best_timestamp = timestamp;
+            }
+        }
+
+        Self {
+            base_instant: best_instant,
+            base_timestamp_us: best_timestamp,
+            last_calibration: best_instant,
+            calibration_interval_secs,
+        }
     }
 
     /// 获取当前时间戳（微秒），使用单调时钟计算，避免系统调用
@@ -54,11 +88,53 @@ impl HighPerformanceClock {
         self.base_timestamp_us + elapsed.as_micros() as i64
     }
 
+    /// 获取高精度当前时间戳（微秒），在必要时进行校准
+    pub fn now_micros_with_calibration(&mut self) -> i64 {
+        // 检查是否需要重新校准
+        if self.last_calibration.elapsed().as_secs() >= self.calibration_interval_secs {
+            self.recalibrate();
+        }
+        self.now_micros()
+    }
+
+    /// 重新校准时钟，减少累积漂移
+    fn recalibrate(&mut self) {
+        let current_monotonic = Instant::now();
+        let current_utc = chrono::Utc::now().timestamp_micros();
+
+        // 计算预期的UTC时间戳（基于单调时钟）
+        let expected_utc = self.base_timestamp_us
+            + current_monotonic.duration_since(self.base_instant).as_micros() as i64;
+
+        // 计算漂移量
+        let drift_us = current_utc - expected_utc;
+
+        // 如果漂移超过1毫秒，进行校准
+        if drift_us.abs() > 1000 {
+            self.base_instant = current_monotonic;
+            self.base_timestamp_us = current_utc;
+        }
+
+        self.last_calibration = current_monotonic;
+    }
+
     /// 计算从指定时间戳到现在的消耗时间（微秒）
     #[inline(always)]
     pub fn elapsed_micros_since(&self, start_timestamp_us: i64) -> i64 {
         self.now_micros() - start_timestamp_us
     }
+
+    /// 获取高精度纳秒时间戳
+    #[inline(always)]
+    pub fn now_nanos(&self) -> i128 {
+        let elapsed = self.base_instant.elapsed();
+        (self.base_timestamp_us as i128 * 1000) + elapsed.as_nanos() as i128
+    }
+
+    /// 重置时钟（强制重新初始化）
+    pub fn reset(&mut self) {
+        *self = Self::new_with_calibration_interval(self.calibration_interval_secs);
+    }
 }
 
 impl Default for HighPerformanceClock {
@@ -67,14 +143,21 @@ impl Default for HighPerformanceClock {
     }
 }
 
-/// 全局高性能时钟实例（使用OnceCell避免重复初始化）
+/// 全局高性能时钟实例
 static HIGH_PERF_CLOCK: once_cell::sync::OnceCell<HighPerformanceClock> =
     once_cell::sync::OnceCell::new();
 
-/// 获取全局高性能时钟实例
+/// 获取全局高性能时钟实例（最简单的实现）
 #[inline(always)]
-pub fn get_high_perf_clock() -> &'static HighPerformanceClock {
-    HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new)
+pub fn get_high_perf_clock() -> i64 {
+    let clock = HIGH_PERF_CLOCK.get_or_init(HighPerformanceClock::new);
+    clock.now_micros()
+}
+
+/// 计算从指定时间戳到现在的消耗时间（微秒）
+#[inline(always)]
+pub fn elapsed_micros_since(start_timestamp_us: i64) -> i64 {
+    get_high_perf_clock() - start_timestamp_us
 }
 
 /// 轻量级事件包装器，避免频繁的Box分配
@@ -147,19 +230,19 @@ pub trait UnifiedEvent: Debug + Send + Sync {
     fn event_type(&self) -> EventType;
 
     /// Get transaction signature
-    fn signature(&self) -> &str;
+    fn signature(&self) -> &Signature;
 
     /// Get slot number
     fn slot(&self) -> u64;
 
     /// Get program received timestamp (milliseconds)
-    fn program_received_time_us(&self) -> i64;
+    fn recv_us(&self) -> i64;
 
     /// Processing time consumption (milliseconds)
-    fn program_handle_time_consuming_us(&self) -> i64;
+    fn handle_us(&self) -> i64;
 
     /// Set processing time consumption (milliseconds)
-    fn set_program_handle_time_consuming_us(&mut self, program_handle_time_consuming_us: i64);
+    fn set_handle_us(&mut self, handle_us: i64);
 
     /// Convert event to Any for downcasting
     fn as_any(&self) -> &dyn std::any::Any;
@@ -182,8 +265,8 @@ pub trait UnifiedEvent: Debug + Send + Sync {
     fn swap_data_is_parsed(&self) -> bool;
 
     /// Get index
-    fn instruction_outer_index(&self) -> i64;
-    fn instruction_inner_index(&self) -> Option<i64>;
+    fn outer_index(&self) -> i64;
+    fn inner_index(&self) -> Option<i64>;
 
     /// Get transaction index in slot
     fn transaction_index(&self) -> Option<u64>;
@@ -202,7 +285,22 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: u64,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
+        outer_index: i64,
+        inner_index: Option<i64>,
+        transaction_index: Option<u64>,
+        config: &GenericEventParseConfig,
+    ) -> Vec<Box<dyn UnifiedEvent>>;
+
+    /// 从内联指令中解析事件数据
+    #[allow(clippy::too_many_arguments)]
+    fn parse_events_from_grpc_inner_instruction(
+        &self,
+        inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction,
+        signature: Signature,
+        slot: u64,
+        block_time: Option<Timestamp>,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         transaction_index: Option<u64>,
@@ -218,7 +316,7 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: u64,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         bot_wallet: Option<Pubkey>,
@@ -227,6 +325,80 @@ pub trait EventParser: Send + Sync {
         callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
     ) -> anyhow::Result<()>;
 
+    /// 从指令中解析事件数据
+    /// TODO: - wait refactor
+    #[allow(clippy::too_many_arguments)]
+    fn parse_events_from_grpc_instruction(
+        &self,
+        instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
+        accounts: &[Pubkey],
+        signature: Signature,
+        slot: u64,
+        block_time: Option<Timestamp>,
+        recv_us: i64,
+        outer_index: i64,
+        inner_index: Option<i64>,
+        bot_wallet: Option<Pubkey>,
+        transaction_index: Option<u64>,
+        inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
+        callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
+    ) -> anyhow::Result<()>;
+
+    #[allow(clippy::too_many_arguments)]
+    async fn parse_instruction_events_from_grpc_transaction(
+        &self,
+        compiled_instructions: &[yellowstone_grpc_proto::prelude::CompiledInstruction],
+        signature: Signature,
+        slot: Option<u64>,
+        block_time: Option<Timestamp>,
+        recv_us: i64,
+        accounts: &[Pubkey],
+        inner_instructions: &[yellowstone_grpc_proto::prelude::InnerInstructions],
+        bot_wallet: Option<Pubkey>,
+        transaction_index: Option<u64>,
+        callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
+    ) -> anyhow::Result<()> {
+        // 获取交易的指令和账户
+        let mut accounts = accounts.to_vec();
+        // 检查交易中是否包含程序
+        let has_program = accounts.iter().any(|account| self.should_handle(account));
+        if has_program {
+            // 解析每个指令
+            for (index, instruction) in compiled_instructions.iter().enumerate() {
+                if let Some(program_id) = accounts.get(instruction.program_id_index as usize) {
+                    if self.should_handle(program_id) {
+                        let max_idx = instruction.accounts.iter().max().unwrap_or(&0);
+                        // 补齐accounts(使用Pubkey::default())
+                        if *max_idx as usize > accounts.len() {
+                            for _i in accounts.len()..*max_idx as usize {
+                                accounts.push(Pubkey::default());
+                            }
+                        }
+                        let inner_instructions = inner_instructions
+                            .iter()
+                            .find(|inner_instruction| inner_instruction.index == index as u32);
+                        self.parse_grpc_instruction(
+                            instruction,
+                            &accounts,
+                            signature,
+                            slot,
+                            block_time,
+                            recv_us,
+                            index as i64,
+                            None,
+                            bot_wallet,
+                            transaction_index,
+                            inner_instructions,
+                            Arc::clone(&callback),
+                        )
+                        .await?;
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// 从VersionedTransaction中解析指令事件的通用方法
     #[allow(clippy::too_many_arguments)]
     async fn parse_instruction_events_from_versioned_transaction(
@@ -235,7 +407,7 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         accounts: &[Pubkey],
         inner_instructions: &[InnerInstructions],
         bot_wallet: Option<Pubkey>,
@@ -268,7 +440,7 @@ pub trait EventParser: Send + Sync {
                             signature,
                             slot,
                             block_time,
-                            program_received_time_us,
+                            recv_us,
                             index as i64,
                             None,
                             bot_wallet,
@@ -290,7 +462,7 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         bot_wallet: Option<Pubkey>,
         transaction_index: Option<u64>,
         inner_instructions: &[InnerInstructions],
@@ -305,7 +477,7 @@ pub trait EventParser: Send + Sync {
             signature,
             slot,
             block_time,
-            program_received_time_us,
+            recv_us,
             bot_wallet,
             transaction_index,
             inner_instructions,
@@ -321,7 +493,7 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         bot_wallet: Option<Pubkey>,
         transaction_index: Option<u64>,
         inner_instructions: &[InnerInstructions],
@@ -333,7 +505,7 @@ pub trait EventParser: Send + Sync {
             signature,
             slot,
             block_time,
-            program_received_time_us,
+            recv_us,
             &accounts,
             inner_instructions,
             bot_wallet,
@@ -344,14 +516,13 @@ pub trait EventParser: Send + Sync {
         Ok(())
     }
 
-    /// 解析交易，使用所有权语义的回调以避免不必要的克隆
-    async fn parse_transaction_owned(
+    async fn parse_grpc_transaction_owned(
         &self,
-        tx: TransactionWithStatusMeta,
+        grpc_tx: SubscribeUpdateTransactionInfo,
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         bot_wallet: Option<Pubkey>,
         transaction_index: Option<u64>,
         callback: Arc<dyn Fn(Box<dyn UnifiedEvent>) + Send + Sync>,
@@ -361,12 +532,12 @@ pub trait EventParser: Send + Sync {
             callback(event.clone_boxed());
         });
         // 调用原始方法
-        self.parse_transaction(
-            tx,
+        self.parse_grpc_transaction(
+            grpc_tx,
             signature,
             slot,
             block_time,
-            program_received_time_us,
+            recv_us,
             bot_wallet,
             transaction_index,
             adapter_callback,
@@ -374,71 +545,97 @@ pub trait EventParser: Send + Sync {
         .await
     }
 
-    async fn parse_transaction(
+    async fn parse_grpc_transaction(
         &self,
-        tx: TransactionWithStatusMeta,
+        grpc_tx: SubscribeUpdateTransactionInfo,
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         bot_wallet: Option<Pubkey>,
         transaction_index: Option<u64>,
         callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
     ) -> anyhow::Result<()> {
-        let versioned_tx = tx.get_transaction();
-        let meta = tx.get_status_meta();
-        let mut address_table_lookups: Vec<Pubkey> = vec![];
-        let mut inner_instructions: Vec<InnerInstructions> = vec![];
-        if let Some(meta) = meta {
-            inner_instructions = meta.inner_instructions.unwrap_or_default();
-            address_table_lookups.reserve(
-                meta.loaded_addresses.writable.len() + meta.loaded_addresses.readonly.len(),
-            );
-            address_table_lookups.extend(
-                meta.loaded_addresses.writable.into_iter().chain(meta.loaded_addresses.readonly),
-            );
-        }
-        let mut accounts = Vec::with_capacity(
-            versioned_tx.message.static_account_keys().len() + address_table_lookups.len(),
-        );
-        accounts.extend_from_slice(versioned_tx.message.static_account_keys());
-        accounts.extend(address_table_lookups);
-        // 使用 Arc 包装共享数据，避免不必要的克隆
-        let accounts_arc = Arc::new(accounts);
-        let inner_instructions_arc = Arc::new(inner_instructions);
-        // 解析指令事件
-        self.parse_instruction_events_from_versioned_transaction(
-            &versioned_tx,
-            signature,
-            slot,
-            block_time,
-            program_received_time_us,
-            &accounts_arc,
-            &inner_instructions_arc,
-            bot_wallet,
-            transaction_index,
-            callback.clone(),
-        )
-        .await?;
+        if let Some(transition) = grpc_tx.transaction {
+            if let Some(message) = &transition.message {
+                let mut address_table_lookups: Vec<Vec<u8>> = vec![];
+                let mut inner_instructions: Vec<
+                    yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions,
+                > = vec![];
 
-        // 解析嵌套指令事件
-        for inner_instruction in inner_instructions_arc.iter() {
-            for (index, instruction) in inner_instruction.instructions.iter().enumerate() {
-                self.parse_instruction(
-                    &instruction.instruction,
-                    &accounts_arc,
+                if let Some(meta) = grpc_tx.meta {
+                    inner_instructions = meta.inner_instructions;
+                    address_table_lookups.reserve(
+                        meta.loaded_writable_addresses.len() + meta.loaded_writable_addresses.len(),
+                    );
+                    let loaded_writable_addresses = meta.loaded_writable_addresses;
+                    let loaded_readonly_addresses = meta.loaded_readonly_addresses;
+                    address_table_lookups.extend(
+                        loaded_writable_addresses.into_iter().chain(loaded_readonly_addresses),
+                    );
+                }
+
+                let mut accounts_bytes: Vec<Vec<u8>> =
+                    Vec::with_capacity(message.account_keys.len() + address_table_lookups.len());
+                accounts_bytes.extend_from_slice(&message.account_keys);
+                accounts_bytes.extend(address_table_lookups);
+                // 转换为 Pubkey
+                let accounts: Vec<Pubkey> = accounts_bytes
+                    .iter()
+                    .filter_map(|account| {
+                        if account.len() == 32 {
+                            Some(Pubkey::try_from(account.as_slice()).unwrap_or_default())
+                        } else {
+                            None
+                        }
+                    })
+                    .collect();
+                // 使用 Arc 包装共享数据，避免不必要的克隆
+                let accounts_arc = Arc::new(accounts);
+                let inner_instructions_arc = Arc::new(inner_instructions);
+                // 解析指令事件
+                let instructions = &message.instructions;
+                self.parse_instruction_events_from_grpc_transaction(
+                    &instructions,
                     signature,
                     slot,
                     block_time,
-                    program_received_time_us,
-                    inner_instruction.index as i64,
-                    Some(index as i64),
+                    recv_us,
+                    &accounts_arc,
+                    &inner_instructions_arc,
                     bot_wallet,
                     transaction_index,
-                    Some(&inner_instruction),
                     callback.clone(),
                 )
                 .await?;
+
+                // 解析嵌套指令事件
+                for inner_instruction in inner_instructions_arc.iter() {
+                    for (index, instruction) in inner_instruction.instructions.iter().enumerate() {
+                        let accounts = &instruction.accounts;
+                        let data = &instruction.data;
+                        let instruction = yellowstone_grpc_proto::prelude::CompiledInstruction {
+                            program_id_index: instruction.program_id_index,
+                            accounts: accounts.to_vec(),
+                            data: data.to_vec(),
+                        };
+                        self.parse_grpc_instruction(
+                            &instruction,
+                            &accounts_arc,
+                            signature,
+                            slot,
+                            block_time,
+                            recv_us,
+                            inner_instruction.index as i64,
+                            Some(index as i64),
+                            bot_wallet,
+                            transaction_index,
+                            Some(&inner_instruction),
+                            callback.clone(),
+                        )
+                        .await?;
+                    }
+                }
             }
         }
 
@@ -535,7 +732,7 @@ pub trait EventParser: Send + Sync {
 
         let slot = transaction.slot;
         let block_time = transaction.block_time.map(|t| Timestamp { seconds: t as i64, nanos: 0 });
-        let program_received_time_us = chrono::Utc::now().timestamp_micros();
+        let recv_us = get_high_perf_clock();
         let bot_wallet = None;
         let transaction_index = None;
         // 解析指令事件
@@ -544,7 +741,7 @@ pub trait EventParser: Send + Sync {
             signature,
             Some(slot),
             block_time,
-            program_received_time_us,
+            recv_us,
             &accounts_arc,
             &inner_instructions_arc,
             bot_wallet,
@@ -562,7 +759,7 @@ pub trait EventParser: Send + Sync {
                     signature,
                     Some(slot),
                     block_time,
-                    program_received_time_us,
+                    recv_us,
                     inner_instruction.index as i64,
                     Some(index as i64),
                     bot_wallet,
@@ -583,7 +780,7 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         transaction_index: Option<u64>,
@@ -595,7 +792,7 @@ pub trait EventParser: Send + Sync {
             signature,
             slot,
             block_time,
-            program_received_time_us,
+            recv_us,
             outer_index,
             inner_index,
             transaction_index,
@@ -604,6 +801,39 @@ pub trait EventParser: Send + Sync {
         Ok(events)
     }
 
+    #[allow(clippy::too_many_arguments)]
+    async fn parse_grpc_instruction(
+        &self,
+        instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
+        accounts: &[Pubkey],
+        signature: Signature,
+        slot: Option<u64>,
+        block_time: Option<Timestamp>,
+        recv_us: i64,
+        outer_index: i64,
+        inner_index: Option<i64>,
+        bot_wallet: Option<Pubkey>,
+        transaction_index: Option<u64>,
+        inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
+        callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
+    ) -> anyhow::Result<()> {
+        let slot = slot.unwrap_or(0);
+        self.parse_events_from_grpc_instruction(
+            instruction,
+            accounts,
+            signature,
+            slot,
+            block_time,
+            recv_us,
+            outer_index,
+            inner_index,
+            bot_wallet,
+            transaction_index,
+            inner_instructions,
+            callback,
+        )
+    }
+
     #[allow(clippy::too_many_arguments)]
     async fn parse_instruction(
         &self,
@@ -612,7 +842,7 @@ pub trait EventParser: Send + Sync {
         signature: Signature,
         slot: Option<u64>,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         bot_wallet: Option<Pubkey>,
@@ -627,7 +857,7 @@ pub trait EventParser: Send + Sync {
             signature,
             slot,
             block_time,
-            program_received_time_us,
+            recv_us,
             outer_index,
             inner_index,
             bot_wallet,
@@ -708,17 +938,16 @@ impl GenericEventParser {
         signature: Signature,
         slot: u64,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         transaction_index: Option<u64>,
     ) -> Option<Box<dyn UnifiedEvent>> {
         if let Some(parser) = config.inner_instruction_parser {
-            let signature_str = Cow::Owned(signature.to_string());
             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_str,
+                signature,
                 slot,
                 timestamp.seconds,
                 block_time_ms,
@@ -727,7 +956,7 @@ impl GenericEventParser {
                 config.program_id,
                 outer_index,
                 inner_index,
-                program_received_time_us,
+                recv_us,
                 transaction_index,
             );
             parser(data, metadata)
@@ -746,17 +975,16 @@ impl GenericEventParser {
         signature: Signature,
         slot: u64,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         transaction_index: Option<u64>,
     ) -> Option<Box<dyn UnifiedEvent>> {
         if let Some(parser) = config.instruction_parser {
-            let signature_str = Cow::Owned(signature.to_string());
             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_str,
+                signature,
                 slot,
                 timestamp.seconds,
                 block_time_ms,
@@ -765,7 +993,7 @@ impl GenericEventParser {
                 config.program_id,
                 outer_index,
                 inner_index,
-                program_received_time_us,
+                recv_us,
                 transaction_index,
             );
             parser(data, account_pubkeys, metadata)
@@ -788,7 +1016,7 @@ impl EventParser for GenericEventParser {
         signature: Signature,
         slot: u64,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         transaction_index: Option<u64>,
@@ -806,7 +1034,43 @@ impl EventParser for GenericEventParser {
             signature,
             slot,
             block_time,
-            program_received_time_us,
+            recv_us,
+            outer_index,
+            inner_index,
+            transaction_index,
+        ) {
+            events.push(event);
+        }
+        events
+    }
+
+    /// 从内联指令中解析事件数据
+    #[allow(clippy::too_many_arguments)]
+    fn parse_events_from_grpc_inner_instruction(
+        &self,
+        inner_instruction: &yellowstone_grpc_proto::prelude::InnerInstruction,
+        signature: Signature,
+        slot: u64,
+        block_time: Option<Timestamp>,
+        recv_us: i64,
+        outer_index: i64,
+        inner_index: Option<i64>,
+        transaction_index: Option<u64>,
+        config: &GenericEventParseConfig,
+    ) -> Vec<Box<dyn UnifiedEvent>> {
+        // Use SIMD-optimized data validation
+        if !SimdUtils::validate_instruction_data_simd(&inner_instruction.data, 16, 0) {
+            return Vec::new();
+        }
+        let data = &inner_instruction.data[16..];
+        let mut events = Vec::new();
+        if let Some(event) = self.parse_inner_instruction_event(
+            config,
+            data,
+            signature,
+            slot,
+            block_time,
+            recv_us,
             outer_index,
             inner_index,
             transaction_index,
@@ -825,7 +1089,7 @@ impl EventParser for GenericEventParser {
         signature: Signature,
         slot: u64,
         block_time: Option<Timestamp>,
-        program_received_time_us: i64,
+        recv_us: i64,
         outer_index: i64,
         inner_index: Option<i64>,
         bot_wallet: Option<Pubkey>,
@@ -877,7 +1141,7 @@ impl EventParser for GenericEventParser {
                     signature,
                     slot,
                     block_time,
-                    program_received_time_us,
+                    recv_us,
                     outer_index,
                     inner_index,
                     transaction_index,
@@ -901,7 +1165,7 @@ impl EventParser for GenericEventParser {
                                 signature,
                                 slot,
                                 block_time,
-                                program_received_time_us,
+                                recv_us,
                                 outer_index,
                                 inner_index,
                                 transaction_index,
@@ -941,9 +1205,144 @@ impl EventParser for GenericEventParser {
                 event.merge(&*inner_instruction_event);
             }
             // 设置处理时间（使用高性能时钟）
-            event.set_program_handle_time_consuming_us(
-                get_high_perf_clock().elapsed_micros_since(program_received_time_us),
-            );
+            event.set_handle_us(elapsed_micros_since(
+                recv_us,
+            ));
+            event = process_event(event, bot_wallet);
+            callback(&event);
+        }
+        Ok(())
+    }
+
+    /// 从指令中解析事件
+    /// TODO: - wait refactor
+    #[allow(clippy::too_many_arguments)]
+    fn parse_events_from_grpc_instruction(
+        &self,
+        instruction: &yellowstone_grpc_proto::prelude::CompiledInstruction,
+        accounts: &[Pubkey],
+        signature: Signature,
+        slot: u64,
+        block_time: Option<Timestamp>,
+        recv_us: i64,
+        outer_index: i64,
+        inner_index: Option<i64>,
+        bot_wallet: Option<Pubkey>,
+        transaction_index: Option<u64>,
+        inner_instructions: Option<&yellowstone_grpc_proto::prelude::InnerInstructions>,
+        callback: Arc<dyn for<'a> Fn(&'a Box<dyn UnifiedEvent>) + Send + Sync>,
+    ) -> anyhow::Result<()> {
+        let program_id = accounts[instruction.program_id_index as usize];
+        if !self.should_handle(&program_id) {
+            return Ok(());
+        }
+        // 一维化并行处理：将所有 (discriminator, config) 组合展开并行处理
+        let all_processing_params: Vec<_> = self
+            .instruction_configs
+            .iter()
+            .filter(|(disc, _)| {
+                // Use SIMD-optimized data validation and discriminator matching
+                SimdUtils::validate_instruction_data_simd(&instruction.data, disc.len(), disc.len())
+                    && SimdUtils::fast_discriminator_match(&instruction.data, disc)
+            })
+            .flat_map(|(disc, configs)| {
+                configs
+                    .iter()
+                    .filter(|config| config.program_id == program_id)
+                    .map(move |config| (disc, config))
+            })
+            .collect();
+
+        // Use SIMD-optimized account indices validation (只需检查一次)
+        if !SimdUtils::validate_account_indices_simd(&instruction.accounts, accounts.len()) {
+            return Ok(());
+        }
+
+        // 使用缓存构建账户公钥列表，避免重复分配 (只需构建一次)
+        let account_pubkeys = {
+            let mut cache_guard = self.account_cache.lock();
+            cache_guard.build_account_pubkeys(&instruction.accounts, accounts).to_vec()
+        };
+
+        // 并行处理所有 (discriminator, config) 组合
+        let all_results: Vec<_> = all_processing_params
+            .iter()
+            .filter_map(|(disc, config)| {
+                let data = &instruction.data[disc.len()..];
+                self.parse_instruction_event(
+                    config,
+                    data,
+                    &account_pubkeys,
+                    signature,
+                    slot,
+                    block_time,
+                    recv_us,
+                    outer_index,
+                    inner_index,
+                    transaction_index,
+                )
+                .map(|event| ((*disc).clone(), (*config).clone(), event))
+            })
+            .collect();
+
+        for (_disc, config, mut event) in all_results {
+            // 阻塞处理：原有的同步逻辑
+            let mut inner_instruction_event: Option<Box<dyn UnifiedEvent>> = None;
+            if inner_instructions.is_some() {
+                let inner_instructions_ref = inner_instructions.unwrap();
+
+                // 并行执行两个任务
+                let (inner_event_result, swap_data_result) = std::thread::scope(|s| {
+                    let inner_event_handle = s.spawn(|| {
+                        for inner_instruction in inner_instructions_ref.instructions.iter() {
+                            let result = self.parse_events_from_grpc_inner_instruction(
+                                &inner_instruction,
+                                signature,
+                                slot,
+                                block_time,
+                                recv_us,
+                                outer_index,
+                                inner_index,
+                                transaction_index,
+                                &config,
+                            );
+                            if result.len() > 0 {
+                                return Some(result[0].clone());
+                            }
+                        }
+                        None
+                    });
+
+                    let swap_data_handle = s.spawn(|| {
+                        if !event.swap_data_is_parsed() {
+                            parse_swap_data_from_next_grpc_instructions(
+                                &*event,
+                                inner_instructions_ref,
+                                inner_index.unwrap_or(-1_i64) as i8,
+                                &accounts,
+                            )
+                        } else {
+                            None
+                        }
+                    });
+
+                    // 等待两个任务完成
+                    (inner_event_handle.join().unwrap(), swap_data_handle.join().unwrap())
+                });
+
+                inner_instruction_event = inner_event_result;
+                if let Some(swap_data) = swap_data_result {
+                    event.set_swap_data(swap_data);
+                }
+            }
+            // 合并事件
+            if let Some(inner_instruction_event) = inner_instruction_event {
+                event.merge(&*inner_instruction_event);
+            }
+            // 设置处理时间（使用高性能时钟）
+            event.set_handle_us(elapsed_micros_since(
+                recv_us,
+            ));
             event = process_event(event, bot_wallet);
             callback(&event);
         }

src/streaming/event_parser/protocols/block/block_meta_event.rs

@@ -1,9 +1,8 @@
-use std::borrow::Cow;
-
 use crate::impl_unified_event;
 use crate::streaming::event_parser::common::{types::EventType, EventMetadata};
 use borsh::BorshDeserialize;
 use serde::{Deserialize, Serialize};
+use solana_sdk::signature::Signature;
 
 /// Block元数据事件
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
@@ -19,10 +18,10 @@ impl BlockMetaEvent {
         slot: u64,
         block_hash: String,
         block_time_ms: i64,
-        program_received_time_us: i64,
+        recv_us: i64,
     ) -> Self {
         let metadata = EventMetadata::new(
-            Cow::Borrowed(""),
+            Signature::default(),
             slot,
             block_time_ms / 1000,
             block_time_ms,
@@ -31,7 +30,7 @@ impl BlockMetaEvent {
             solana_sdk::pubkey::Pubkey::default(),
             0,
             None,
-            program_received_time_us,
+            recv_us,
             None,
         );
         Self { metadata, slot, block_hash }

src/streaming/event_parser/protocols/bonk/events.rs

@@ -237,6 +237,7 @@ impl_unified_event!(
 // Migrate to CP Swap event
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct BonkMigrateToCpswapEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub payer: Pubkey,
     pub base_mint: Pubkey,
@@ -276,10 +277,10 @@ impl_unified_event!(BonkMigrateToCpswapEvent,);
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BonkPoolStateAccountEvent {
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub pool_state: PoolState,
 }
@@ -289,10 +290,10 @@ impl_unified_event!(BonkPoolStateAccountEvent,);
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BonkGlobalConfigAccountEvent {
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub global_config: GlobalConfig,
 }
@@ -302,10 +303,10 @@ impl_unified_event!(BonkGlobalConfigAccountEvent,);
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct BonkPlatformConfigAccountEvent {
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub platform_config: PlatformConfig,
 }

src/streaming/event_parser/protocols/bonk/types.rs

@@ -142,10 +142,10 @@ pub fn pool_state_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             pool_state,
         }))
@@ -193,10 +193,10 @@ pub fn global_config_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             global_config,
         }))
@@ -241,10 +241,10 @@ pub fn platform_config_parser(
     {
         Some(Box::new(BonkPlatformConfigAccountEvent {
             metadata,
-            pubkey: account.pubkey.to_string(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             platform_config,
         }))

src/streaming/event_parser/protocols/pumpfun/events.rs

@@ -228,10 +228,10 @@ impl_unified_event!(
 pub struct PumpFunBondingCurveAccountEvent {
     #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub bonding_curve: BondingCurve,
 }
@@ -243,10 +243,10 @@ impl_unified_event!(PumpFunBondingCurveAccountEvent,);
 pub struct PumpFunGlobalAccountEvent {
     #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub global: Global,
 }

src/streaming/event_parser/protocols/pumpfun/types.rs

@@ -41,10 +41,10 @@ pub fn bonding_curve_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             bonding_curve,
         }))
@@ -91,10 +91,10 @@ pub fn global_parser(
     if let Some(global) = global_decode(&account.data[8..GLOBAL_SIZE + 8]) {
         Some(Box::new(PumpFunGlobalAccountEvent {
             metadata,
-            pubkey: account.pubkey.to_string(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             global,
         }))

src/streaming/event_parser/protocols/pumpswap/events.rs

@@ -366,11 +366,12 @@ impl_unified_event!(
 /// 全局配置
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct PumpSwapGlobalConfigAccountEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub global_config: GlobalConfig,
 }
@@ -379,11 +380,12 @@ impl_unified_event!(PumpSwapGlobalConfigAccountEvent,);
 /// 池
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct PumpSwapPoolAccountEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub pool: Pool,
 }

src/streaming/event_parser/protocols/pumpswap/types.rs

@@ -43,10 +43,10 @@ pub fn global_config_parser(
     if let Some(config) = global_config_decode(&account.data[8..GLOBAL_CONFIG_SIZE + 8]) {
         Some(Box::new(PumpSwapGlobalConfigAccountEvent {
             metadata,
-            pubkey: account.pubkey.to_string(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             global_config: config,
         }))
@@ -88,10 +88,10 @@ pub fn pool_parser(
     if let Some(pool) = pool_decode(&account.data[8..POOL_SIZE + 8]) {
         Some(Box::new(PumpSwapPoolAccountEvent {
             metadata,
-            pubkey: account.pubkey.to_string(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             pool: pool,
         }))

src/streaming/event_parser/protocols/raydium_amm_v4/events.rs

@@ -9,6 +9,7 @@ use solana_sdk::pubkey::Pubkey;
 /// 交易
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumAmmV4SwapEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     // base in
     pub amount_in: u64,
@@ -42,6 +43,7 @@ impl_unified_event!(RaydiumAmmV4SwapEvent,);
 /// 添加流动性
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumAmmV4DepositEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub max_coin_amount: u64,
     pub max_pc_amount: u64,
@@ -67,6 +69,7 @@ impl_unified_event!(RaydiumAmmV4DepositEvent,);
 /// 初始化
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumAmmV4Initialize2Event {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub nonce: u8,
     pub open_time: u64,
@@ -100,6 +103,7 @@ impl_unified_event!(RaydiumAmmV4Initialize2Event,);
 /// 移除流动性
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumAmmV4WithdrawEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub amount: u64,
 
@@ -131,6 +135,7 @@ impl_unified_event!(RaydiumAmmV4WithdrawEvent,);
 /// 提现
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumAmmV4WithdrawPnlEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
 
     pub token_program: Pubkey,
@@ -156,11 +161,12 @@ impl_unified_event!(RaydiumAmmV4WithdrawPnlEvent,);
 /// 池信息
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumAmmV4AmmInfoAccountEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub amm_info: AmmInfo,
 }

src/streaming/event_parser/protocols/raydium_amm_v4/types.rs

@@ -96,10 +96,10 @@ pub fn amm_info_parser(
     if let Some(amm_info) = amm_info_decode(&account.data[..AMM_INFO_SIZE]) {
         Some(Box::new(RaydiumAmmV4AmmInfoAccountEvent {
             metadata,
-            pubkey: account.pubkey.to_string(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             amm_info: amm_info,
         }))

src/streaming/event_parser/protocols/raydium_clmm/events.rs

@@ -223,10 +223,10 @@ impl_unified_event!(RaydiumClmmOpenPositionV2Event,);
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RaydiumClmmAmmConfigAccountEvent {
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub amm_config: AmmConfig,
 }
@@ -236,10 +236,10 @@ impl_unified_event!(RaydiumClmmAmmConfigAccountEvent,);
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RaydiumClmmPoolStateAccountEvent {
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub pool_state: PoolState,
 }
@@ -249,10 +249,10 @@ impl_unified_event!(RaydiumClmmPoolStateAccountEvent,);
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct RaydiumClmmTickArrayStateAccountEvent {
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub tick_array_state: TickArrayState,
 }

src/streaming/event_parser/protocols/raydium_clmm/types.rs

@@ -47,10 +47,10 @@ pub fn amm_config_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             amm_config: amm_config,
         }))
@@ -135,10 +135,10 @@ pub fn pool_state_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             pool_state: pool_state,
         }))
@@ -218,10 +218,10 @@ pub fn tick_array_state_parser(
     {
         Some(Box::new(RaydiumClmmTickArrayStateAccountEvent {
             metadata,
-            pubkey: account.pubkey.to_string(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             tick_array_state: tick_array_state,
         }))

src/streaming/event_parser/protocols/raydium_cpmm/events.rs

@@ -10,6 +10,7 @@ use solana_sdk::pubkey::Pubkey;
 /// 交易
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumCpmmSwapEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub amount_in: u64,
     pub minimum_amount_out: u64,
@@ -35,6 +36,7 @@ impl_unified_event!(RaydiumCpmmSwapEvent,);
 /// 存款
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumCpmmDepositEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub lp_token_amount: u64,
     pub maximum_token0_amount: u64,
@@ -59,6 +61,7 @@ impl_unified_event!(RaydiumCpmmDepositEvent,);
 /// 初始化
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumCpmmInitializeEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub init_amount0: u64,
     pub init_amount1: u64,
@@ -90,6 +93,7 @@ impl_unified_event!(RaydiumCpmmInitializeEvent,);
 /// 提款
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumCpmmWithdrawEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
     pub lp_token_amount: u64,
     pub minimum_token0_amount: u64,
@@ -115,11 +119,12 @@ impl_unified_event!(RaydiumCpmmWithdrawEvent,);
 /// 池配置
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumCpmmAmmConfigAccountEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub amm_config: AmmConfig,
 }
@@ -128,11 +133,12 @@ impl_unified_event!(RaydiumCpmmAmmConfigAccountEvent,);
 /// 池状态
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)]
 pub struct RaydiumCpmmPoolStateAccountEvent {
+    #[borsh(skip)]
     pub metadata: EventMetadata,
-    pub pubkey: String,
+    pub pubkey: Pubkey,
     pub executable: bool,
     pub lamports: u64,
-    pub owner: String,
+    pub owner: Pubkey,
     pub rent_epoch: u64,
     pub pool_state: PoolState,
 }

src/streaming/event_parser/protocols/raydium_cpmm/types.rs

@@ -46,10 +46,10 @@ pub fn amm_config_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             amm_config: amm_config,
         }))
@@ -104,10 +104,10 @@ pub fn pool_state_parser(
     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(),
+            pubkey: account.pubkey,
             executable: account.executable,
             lamports: account.lamports,
-            owner: account.owner.to_string(),
+            owner: account.owner,
             rent_epoch: account.rent_epoch,
             pool_state: pool_state,
         }))

src/streaming/grpc/mod.rs

@@ -1,10 +1,12 @@
 // gRPC 相关模块
 pub mod connection;
+pub mod pool;
 pub mod subscription;
 pub mod types;
 
 // 重新导出主要类型
 pub use connection::*;
+pub use pool::*;
 pub use subscription::*;
 pub use types::*;
 

src/streaming/grpc/pool.rs

@@ -0,0 +1,437 @@
+use solana_sdk::{pubkey::Pubkey, signature::Signature};
+use std::collections::VecDeque;
+use std::ops::DerefMut;
+use std::sync::{Arc, Mutex};
+use yellowstone_grpc_proto::{
+    geyser::{SubscribeUpdateAccount, SubscribeUpdateBlockMeta, SubscribeUpdateTransaction},
+    prost_types::Timestamp,
+};
+
+use super::types::{AccountPretty, BlockMetaPretty, TransactionPretty};
+use crate::streaming::event_parser::core::traits::get_high_perf_clock;
+
+/// 通用对象池特征
+pub trait ObjectPool<T> {
+    fn acquire(&self) -> PooledObject<T>;
+    fn return_object(&self, obj: Box<T>);
+}
+
+/// 带自动归还的智能指针
+pub struct PooledObject<T> {
+    object: Option<Box<T>>,
+    pool: Arc<Mutex<VecDeque<Box<T>>>>,
+    max_size: usize,
+}
+
+impl<T> PooledObject<T> {
+    fn new(object: Box<T>, pool: Arc<Mutex<VecDeque<Box<T>>>>, max_size: usize) -> Self {
+        Self { object: Some(object), pool, max_size }
+    }
+}
+
+impl<T> Drop for PooledObject<T> {
+    fn drop(&mut self) {
+        if let Some(obj) = self.object.take() {
+            let mut pool = self.pool.lock().unwrap();
+            if pool.len() < self.max_size {
+                pool.push_back(obj);
+            }
+            // 超过最大容量时直接丢弃
+        }
+    }
+}
+
+impl<T> std::ops::Deref for PooledObject<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        self.object.as_ref().unwrap()
+    }
+}
+
+impl<T> std::ops::DerefMut for PooledObject<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.object.as_mut().unwrap()
+    }
+}
+
+/// AccountPretty 对象池
+pub struct AccountPrettyPool {
+    pool: Arc<Mutex<VecDeque<Box<AccountPretty>>>>,
+    max_size: usize,
+}
+
+impl AccountPrettyPool {
+    pub fn new(initial_size: usize, max_size: usize) -> Self {
+        let mut pool = VecDeque::with_capacity(initial_size);
+
+        // 预分配对象
+        for _ in 0..initial_size {
+            pool.push_back(Box::new(AccountPretty::default()));
+        }
+
+        Self { pool: Arc::new(Mutex::new(pool)), max_size }
+    }
+
+    pub fn acquire(&self) -> PooledAccountPretty {
+        let mut pool = self.pool.lock().unwrap();
+        let account = match pool.pop_front() {
+            Some(reused) => reused,
+            None => Box::new(AccountPretty::default()),
+        };
+
+        PooledAccountPretty { account, pool: Arc::clone(&self.pool), max_size: self.max_size }
+    }
+}
+
+/// 带自动归还的 AccountPretty
+pub struct PooledAccountPretty {
+    account: Box<AccountPretty>,
+    pool: Arc<Mutex<VecDeque<Box<AccountPretty>>>>,
+    max_size: usize,
+}
+
+impl PooledAccountPretty {
+    /// 从 gRPC 更新重置数据
+    pub fn reset_from_update(&mut self, account_update: SubscribeUpdateAccount) {
+        let account_info = account_update.account.unwrap();
+
+        self.account.slot = account_update.slot;
+        self.account.signature = if let Some(txn_signature) = account_info.txn_signature {
+            Signature::try_from(txn_signature.as_slice()).expect("valid signature")
+        } else {
+            Signature::default()
+        };
+        self.account.pubkey =
+            Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey");
+        self.account.executable = account_info.executable;
+        self.account.lamports = account_info.lamports;
+        self.account.owner = Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey");
+        self.account.rent_epoch = account_info.rent_epoch;
+
+        // 优化数据字段的重用
+        let new_data = account_info.data;
+        if self.account.data.capacity() >= new_data.len() {
+            self.account.data.clear();
+            self.account.data.extend_from_slice(&new_data);
+        } else {
+            self.account.data = new_data;
+        }
+
+        self.account.recv_us = get_high_perf_clock();
+    }
+}
+
+impl Drop for PooledAccountPretty {
+    fn drop(&mut self) {
+        let mut pool = self.pool.lock().unwrap();
+        if pool.len() < self.max_size {
+            // 清理敏感数据
+            self.account.data.clear();
+            self.account.signature = Signature::default();
+            self.account.pubkey = Pubkey::default();
+            self.account.owner = Pubkey::default();
+            pool.push_back(std::mem::take(&mut self.account));
+        }
+    }
+}
+
+impl std::ops::Deref for PooledAccountPretty {
+    type Target = AccountPretty;
+
+    fn deref(&self) -> &Self::Target {
+        &self.account
+    }
+}
+
+impl std::ops::DerefMut for PooledAccountPretty {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.account
+    }
+}
+
+/// BlockMetaPretty 对象池
+pub struct BlockMetaPrettyPool {
+    pool: Arc<Mutex<VecDeque<Box<BlockMetaPretty>>>>,
+    max_size: usize,
+}
+
+impl BlockMetaPrettyPool {
+    pub fn new(initial_size: usize, max_size: usize) -> Self {
+        let mut pool = VecDeque::with_capacity(initial_size);
+
+        // 预分配对象
+        for _ in 0..initial_size {
+            pool.push_back(Box::new(BlockMetaPretty::default()));
+        }
+
+        Self { pool: Arc::new(Mutex::new(pool)), max_size }
+    }
+
+    pub fn acquire(&self) -> PooledBlockMetaPretty {
+        let mut pool = self.pool.lock().unwrap();
+        let block_meta = match pool.pop_front() {
+            Some(reused) => reused,
+            None => Box::new(BlockMetaPretty::default()),
+        };
+
+        PooledBlockMetaPretty { block_meta, pool: Arc::clone(&self.pool), max_size: self.max_size }
+    }
+}
+
+/// 带自动归还的 BlockMetaPretty
+pub struct PooledBlockMetaPretty {
+    block_meta: Box<BlockMetaPretty>,
+    pool: Arc<Mutex<VecDeque<Box<BlockMetaPretty>>>>,
+    max_size: usize,
+}
+
+impl PooledBlockMetaPretty {
+    /// 从 gRPC 更新重置数据
+    pub fn reset_from_update(
+        &mut self,
+        block_update: SubscribeUpdateBlockMeta,
+        block_time: Option<Timestamp>,
+    ) {
+        self.block_meta.slot = block_update.slot;
+        self.block_meta.block_hash = block_update.blockhash;
+        self.block_meta.block_time = block_time;
+        self.block_meta.recv_us = get_high_perf_clock();
+    }
+}
+
+impl Drop for PooledBlockMetaPretty {
+    fn drop(&mut self) {
+        let mut pool = self.pool.lock().unwrap();
+        if pool.len() < self.max_size {
+            // 清理数据
+            self.block_meta.block_hash.clear();
+            self.block_meta.block_time = None;
+            pool.push_back(std::mem::take(&mut self.block_meta));
+        }
+    }
+}
+
+impl std::ops::Deref for PooledBlockMetaPretty {
+    type Target = BlockMetaPretty;
+
+    fn deref(&self) -> &Self::Target {
+        &self.block_meta
+    }
+}
+
+impl std::ops::DerefMut for PooledBlockMetaPretty {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.block_meta
+    }
+}
+
+/// TransactionPretty 对象池
+pub struct TransactionPrettyPool {
+    pool: Arc<Mutex<VecDeque<Box<TransactionPretty>>>>,
+    max_size: usize,
+}
+
+impl TransactionPrettyPool {
+    pub fn new(initial_size: usize, max_size: usize) -> Self {
+        let mut pool = VecDeque::with_capacity(initial_size);
+
+        // 预分配对象
+        for _ in 0..initial_size {
+            pool.push_back(Box::new(TransactionPretty::default()));
+        }
+
+        Self { pool: Arc::new(Mutex::new(pool)), max_size }
+    }
+
+    pub fn acquire(&self) -> PooledTransactionPretty {
+        let mut pool = self.pool.lock().unwrap();
+        let transaction = match pool.pop_front() {
+            Some(reused) => reused,
+            None => Box::new(TransactionPretty::default()),
+        };
+
+        PooledTransactionPretty {
+            transaction,
+            pool: Arc::clone(&self.pool),
+            max_size: self.max_size,
+        }
+    }
+}
+
+/// 带自动归还的 TransactionPretty
+pub struct PooledTransactionPretty {
+    transaction: Box<TransactionPretty>,
+    pool: Arc<Mutex<VecDeque<Box<TransactionPretty>>>>,
+    max_size: usize,
+}
+
+impl PooledTransactionPretty {
+    /// 从 gRPC 更新重置数据
+    pub fn reset_from_update(
+        &mut self,
+        tx_update: SubscribeUpdateTransaction,
+        block_time: Option<Timestamp>,
+    ) {
+        let tx = tx_update.transaction.expect("should be defined");
+
+        self.transaction.slot = tx_update.slot;
+        self.transaction.transaction_index = Some(tx.index);
+        self.transaction.block_time = block_time;
+        self.transaction.block_hash.clear(); // 重置 block_hash
+        self.transaction.signature =
+            Signature::try_from(tx.signature.as_slice()).expect("valid signature");
+        self.transaction.is_vote = tx.is_vote;
+        self.transaction.recv_us = get_high_perf_clock();
+        self.transaction.grpc_tx = tx;
+    }
+}
+
+impl Drop for PooledTransactionPretty {
+    fn drop(&mut self) {
+        let mut pool = self.pool.lock().unwrap();
+        if pool.len() < self.max_size {
+            // 清理数据
+            self.transaction.block_hash.clear();
+            self.transaction.block_time = None;
+            self.transaction.signature = Signature::default();
+            pool.push_back(std::mem::take(&mut self.transaction));
+        }
+    }
+}
+
+impl std::ops::Deref for PooledTransactionPretty {
+    type Target = TransactionPretty;
+
+    fn deref(&self) -> &Self::Target {
+        &self.transaction
+    }
+}
+
+impl std::ops::DerefMut for PooledTransactionPretty {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.transaction
+    }
+}
+
+/// EventPretty 对象池（组合池）
+pub struct EventPrettyPool {
+    account_pool: AccountPrettyPool,
+    block_pool: BlockMetaPrettyPool,
+    transaction_pool: TransactionPrettyPool,
+}
+
+impl EventPrettyPool {
+    pub fn new() -> Self {
+        Self {
+            account_pool: AccountPrettyPool::new(10000, 20000),
+            block_pool: BlockMetaPrettyPool::new(500, 1000),
+            transaction_pool: TransactionPrettyPool::new(10000, 20000),
+        }
+    }
+
+    /// 获取账户事件对象
+    pub fn acquire_account(&self) -> PooledAccountPretty {
+        self.account_pool.acquire()
+    }
+
+    /// 获取区块事件对象
+    pub fn acquire_block(&self) -> PooledBlockMetaPretty {
+        self.block_pool.acquire()
+    }
+
+    /// 获取交易事件对象
+    pub fn acquire_transaction(&self) -> PooledTransactionPretty {
+        self.transaction_pool.acquire()
+    }
+}
+
+/// 对象池管理器（单例）
+pub struct PoolManager {
+    event_pool: EventPrettyPool,
+}
+
+impl PoolManager {
+    pub fn new() -> Self {
+        Self { event_pool: EventPrettyPool::new() }
+    }
+
+    pub fn get_event_pool(&self) -> &EventPrettyPool {
+        &self.event_pool
+    }
+}
+
+impl Default for PoolManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// 工厂函数用于创建优化的 EventPretty
+impl EventPrettyPool {
+    /// 创建账户事件 - 使用对象池优化
+    pub fn create_account_event_optimized(&self, update: SubscribeUpdateAccount) -> AccountPretty {
+        let mut pooled_account = self.acquire_account();
+        pooled_account.reset_from_update(update);
+        // 移动数据而不是克隆，避免多余的内存分配
+        let result = std::mem::replace(pooled_account.deref_mut(), AccountPretty::default());
+        result
+    }
+
+    /// 创建区块事件 - 使用对象池优化
+    pub fn create_block_event_optimized(
+        &self,
+        update: SubscribeUpdateBlockMeta,
+        block_time: Option<Timestamp>,
+    ) -> BlockMetaPretty {
+        let mut pooled_block = self.acquire_block();
+        pooled_block.reset_from_update(update, block_time);
+        // 移动数据而不是克隆
+        let result = std::mem::replace(pooled_block.deref_mut(), BlockMetaPretty::default());
+        result
+    }
+
+    /// 创建交易事件 - 使用对象池优化
+    pub fn create_transaction_event_optimized(
+        &self,
+        update: SubscribeUpdateTransaction,
+        block_time: Option<Timestamp>,
+    ) -> TransactionPretty {
+        let mut pooled_tx = self.acquire_transaction();
+        pooled_tx.reset_from_update(update, block_time);
+        // 移动数据而不是克隆
+        let result = std::mem::replace(pooled_tx.deref_mut(), TransactionPretty::default());
+        result
+    }
+}
+
+// 全局池管理器实例
+lazy_static::lazy_static! {
+    pub static ref GLOBAL_POOL_MANAGER: PoolManager = PoolManager::new();
+}
+
+/// 便捷的全局工厂函数
+pub mod factory {
+    use super::*;
+
+    /// 使用对象池创建账户事件（推荐用于高性能场景）
+    pub fn create_account_pretty_pooled(update: SubscribeUpdateAccount) -> AccountPretty {
+        GLOBAL_POOL_MANAGER.get_event_pool().create_account_event_optimized(update)
+    }
+
+    /// 使用对象池创建区块事件（推荐用于高性能场景）
+    pub fn create_block_meta_pretty_pooled(
+        update: SubscribeUpdateBlockMeta,
+        block_time: Option<Timestamp>,
+    ) -> BlockMetaPretty {
+        GLOBAL_POOL_MANAGER.get_event_pool().create_block_event_optimized(update, block_time)
+    }
+
+    /// 使用对象池创建交易事件（推荐用于高性能场景）
+    pub fn create_transaction_pretty_pooled(
+        update: SubscribeUpdateTransaction,
+        block_time: Option<Timestamp>,
+    ) -> TransactionPretty {
+        GLOBAL_POOL_MANAGER.get_event_pool().create_transaction_event_optimized(update, block_time)
+    }
+}

src/streaming/grpc/types.rs

@@ -1,10 +1,10 @@
 use solana_sdk::{pubkey::Pubkey, signature::Signature};
-use solana_transaction_status::TransactionWithStatusMeta;
+use solana_transaction_status::{TransactionWithStatusMeta, VersionedTransactionWithStatusMeta};
 use std::{collections::HashMap, fmt};
 use yellowstone_grpc_proto::{
     geyser::{
-        SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions, SubscribeUpdateAccount,
-        SubscribeUpdateBlockMeta, SubscribeUpdateTransaction,
+        SubscribeRequestFilterAccounts, SubscribeRequestFilterTransactions,
+        SubscribeUpdateTransactionInfo,
     },
     prost_types::Timestamp,
 };
@@ -12,14 +12,14 @@ use yellowstone_grpc_proto::{
 pub type TransactionsFilterMap = HashMap<String, SubscribeRequestFilterTransactions>;
 pub type AccountsFilterMap = HashMap<String, SubscribeRequestFilterAccounts>;
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub enum EventPretty {
     BlockMeta(BlockMetaPretty),
     Transaction(TransactionPretty),
     Account(AccountPretty),
 }
 
-#[derive(Clone)]
+#[derive(Clone, Default)]
 pub struct AccountPretty {
     pub slot: u64,
     pub signature: Signature,
@@ -29,7 +29,7 @@ pub struct AccountPretty {
     pub owner: Pubkey,
     pub rent_epoch: u64,
     pub data: Vec<u8>,
-    pub program_received_time_us: i64,
+    pub recv_us: i64,
 }
 
 impl fmt::Debug for AccountPretty {
@@ -47,12 +47,12 @@ impl fmt::Debug for AccountPretty {
     }
 }
 
-#[derive(Clone)]
+#[derive(Clone, Default)]
 pub struct BlockMetaPretty {
     pub slot: u64,
     pub block_hash: String,
     pub block_time: Option<Timestamp>,
-    pub program_received_time_us: i64,
+    pub recv_us: i64,
 }
 
 impl fmt::Debug for BlockMetaPretty {
@@ -61,7 +61,7 @@ impl fmt::Debug for BlockMetaPretty {
             .field("slot", &self.slot)
             .field("block_hash", &self.block_hash)
             .field("block_time", &self.block_time)
-            .field("program_received_time_us", &self.program_received_time_us)
+            .field("recv_us", &self.recv_us)
             .finish()
     }
 }
@@ -74,8 +74,8 @@ pub struct TransactionPretty {
     pub block_time: Option<Timestamp>,
     pub signature: Signature,
     pub is_vote: bool,
-    pub tx: TransactionWithStatusMeta,
-    pub program_received_time_us: i64,
+    pub recv_us: i64,
+    pub grpc_tx: SubscribeUpdateTransactionInfo,
 }
 
 impl fmt::Debug for TransactionPretty {
@@ -85,68 +85,83 @@ impl fmt::Debug for TransactionPretty {
             .field("transaction_index", &self.transaction_index)
             .field("signature", &self.signature)
             .field("is_vote", &self.is_vote)
-            .field("program_received_time_us", &self.program_received_time_us)
+            .field("recv_us", &self.recv_us)
             .finish()
     }
 }
 
-impl From<SubscribeUpdateAccount> for AccountPretty {
-    fn from(account: SubscribeUpdateAccount) -> Self {
-        let account_info = account.account.unwrap();
+impl Default for TransactionPretty {
+    fn default() -> Self {
         Self {
-            slot: account.slot,
-            signature: if let Some(txn_signature) = account_info.txn_signature {
-                Signature::try_from(txn_signature.as_slice()).expect("valid signature")
-            } else {
-                Signature::default()
-            },
-            pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"),
-            executable: account_info.executable,
-            lamports: account_info.lamports,
-            owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"),
-            rent_epoch: account_info.rent_epoch,
-            data: account_info.data,
-            program_received_time_us: chrono::Utc::now().timestamp_micros(),
+            slot: 0,
+            transaction_index: None,
+            block_hash: String::new(),
+            block_time: None,
+            signature: Signature::default(),
+            is_vote: false,
+            grpc_tx: SubscribeUpdateTransactionInfo::default(),
+            recv_us: 0,
         }
     }
 }
 
-impl From<(SubscribeUpdateBlockMeta, Option<Timestamp>)> for BlockMetaPretty {
-    fn from(
-        (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (
-            SubscribeUpdateBlockMeta,
-            Option<Timestamp>,
-        ),
-    ) -> Self {
-        Self {
-            block_hash: blockhash.to_string(),
-            block_time,
-            slot,
-            program_received_time_us: chrono::Utc::now().timestamp_micros(),
-        }
-    }
-}
+// impl From<SubscribeUpdateAccount> for AccountPretty {
+//     fn from(account: SubscribeUpdateAccount) -> Self {
+//         let account_info = account.account.unwrap();
+//         Self {
+//             slot: account.slot,
+//             signature: if let Some(txn_signature) = account_info.txn_signature {
+//                 Signature::try_from(txn_signature.as_slice()).expect("valid signature")
+//             } else {
+//                 Signature::default()
+//             },
+//             pubkey: Pubkey::try_from(account_info.pubkey.as_slice()).expect("valid pubkey"),
+//             executable: account_info.executable,
+//             lamports: account_info.lamports,
+//             owner: Pubkey::try_from(account_info.owner.as_slice()).expect("valid pubkey"),
+//             rent_epoch: account_info.rent_epoch,
+//             data: account_info.data,
+//             recv_us: get_high_perf_clock(),
+//         }
+//     }
+// }
 
-impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
-    fn from(
-        (SubscribeUpdateTransaction { transaction, slot }, block_time): (
-            SubscribeUpdateTransaction,
-            Option<Timestamp>,
-        ),
-    ) -> Self {
-        let tx = transaction.expect("should be defined");
-        // 根据用户说明，交易索引在 transaction.index 中
-        let transaction_index = tx.index;
-        Self {
-            slot,
-            transaction_index: Some(transaction_index), // 提取交易索引
-            block_time,
-            block_hash: "".to_string(),
-            signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
-            is_vote: tx.is_vote,
-            tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx)
-                .expect("valid tx with meta"),
-            program_received_time_us: chrono::Utc::now().timestamp_micros(),
-        }
-    }
-}
+// impl From<(SubscribeUpdateBlockMeta, Option<Timestamp>)> for BlockMetaPretty {
+//     fn from(
+//         (SubscribeUpdateBlockMeta { slot, blockhash, .. }, block_time): (
+//             SubscribeUpdateBlockMeta,
+//             Option<Timestamp>,
+//         ),
+//     ) -> Self {
+//         Self {
+//             block_hash: blockhash,
+//             block_time,
+//             slot,
+//             recv_us: get_high_perf_clock(),
+//         }
+//     }
+// }
+
+// impl From<(SubscribeUpdateTransaction, Option<Timestamp>)> for TransactionPretty {
+//     fn from(
+//         (SubscribeUpdateTransaction { transaction, slot }, block_time): (
+//             SubscribeUpdateTransaction,
+//             Option<Timestamp>,
+//         ),
+//     ) -> Self {
+//         let tx = transaction.expect("should be defined");
+//         // 根据用户说明，交易索引在 transaction.index 中
+//         let transaction_index = tx.index;
+//         Self {
+//             slot,
+//             transaction_index: Some(transaction_index), // 提取交易索引
+//             block_time,
+//             block_hash: String::new(),
+//             signature: Signature::try_from(tx.signature.as_slice()).expect("valid signature"),
+//             is_vote: tx.is_vote,
+//             tx: yellowstone_grpc_proto::convert_from::create_tx_with_meta(tx)
+//                 .expect("valid tx with meta"),
+//             recv_us: get_high_perf_clock(),
+//         }
+//     }
+// }

src/streaming/shred/mod.rs

@@ -1,9 +1,11 @@
 // ShredStream 相关模块
 pub mod connection;
+pub mod pool;
 pub mod types;
 
 // 重新导出主要类型
 pub use connection::*;
+pub use pool::*;
 pub use types::*;
 
 // 从公用模块重新导出

src/streaming/shred/pool.rs

@@ -0,0 +1,156 @@
+use std::sync::{Arc, Mutex};
+use std::collections::VecDeque;
+use std::ops::DerefMut;
+use solana_sdk::transaction::VersionedTransaction;
+
+use super::TransactionWithSlot;
+
+
+/// TransactionWithSlot 对象池
+pub struct TransactionWithSlotPool {
+    pool: Arc<Mutex<VecDeque<Box<TransactionWithSlot>>>>,
+    max_size: usize,
+}
+
+impl TransactionWithSlotPool {
+    pub fn new(initial_size: usize, max_size: usize) -> Self {
+        let mut pool = VecDeque::with_capacity(initial_size);
+
+        // 预分配对象
+        for _ in 0..initial_size {
+            pool.push_back(Box::new(TransactionWithSlot::default()));
+        }
+
+        Self { pool: Arc::new(Mutex::new(pool)), max_size }
+    }
+
+    pub fn acquire(&self) -> PooledTransactionWithSlot {
+        let mut pool = self.pool.lock().unwrap();
+        let transaction = match pool.pop_front() {
+            Some(reused) => reused,
+            None => Box::new(TransactionWithSlot::default()),
+        };
+
+        PooledTransactionWithSlot { 
+            transaction, 
+            pool: Arc::clone(&self.pool), 
+            max_size: self.max_size 
+        }
+    }
+}
+
+/// 带自动归还的 TransactionWithSlot
+pub struct PooledTransactionWithSlot {
+    transaction: Box<TransactionWithSlot>,
+    pool: Arc<Mutex<VecDeque<Box<TransactionWithSlot>>>>,
+    max_size: usize,
+}
+
+impl PooledTransactionWithSlot {
+    /// 从原始数据重置
+    pub fn reset_from_data(
+        &mut self, 
+        transaction: VersionedTransaction, 
+        slot: u64, 
+        recv_us: i64
+    ) {
+        self.transaction.transaction = transaction;
+        self.transaction.slot = slot;
+        self.transaction.recv_us = recv_us;
+    }
+
+    /// 使用优化的工厂方法创建 TransactionWithSlot（移动数据而不是克隆）
+    pub fn into_transaction_with_slot(mut self) -> TransactionWithSlot {
+        // 移动数据而不是克隆，避免多余的内存分配
+        std::mem::replace(self.deref_mut(), TransactionWithSlot::default())
+    }
+}
+
+impl Drop for PooledTransactionWithSlot {
+    fn drop(&mut self) {
+        let mut pool = self.pool.lock().unwrap();
+        if pool.len() < self.max_size {
+            // 清理敏感数据
+            self.transaction.slot = 0;
+            self.transaction.recv_us = 0;
+            // 重置交易为默认值以清理敏感数据
+            self.transaction.transaction = VersionedTransaction::default();
+            pool.push_back(std::mem::take(&mut self.transaction));
+        }
+    }
+}
+
+impl std::ops::Deref for PooledTransactionWithSlot {
+    type Target = TransactionWithSlot;
+
+    fn deref(&self) -> &Self::Target {
+        &self.transaction
+    }
+}
+
+impl std::ops::DerefMut for PooledTransactionWithSlot {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.transaction
+    }
+}
+
+/// Shred 对象池管理器
+pub struct ShredPoolManager {
+    transaction_pool: TransactionWithSlotPool,
+}
+
+impl ShredPoolManager {
+    pub fn new() -> Self {
+        Self {
+            transaction_pool: TransactionWithSlotPool::new(
+                5000,  // 初始大小 - Shred 事件通常较多
+                15000, // 最大大小
+            ),
+        }
+    }
+
+    pub fn get_transaction_pool(&self) -> &TransactionWithSlotPool {
+        &self.transaction_pool
+    }
+
+    /// 创建优化的 TransactionWithSlot
+    pub fn create_transaction_with_slot_optimized(
+        &self,
+        transaction: VersionedTransaction,
+        slot: u64,
+        recv_us: i64,
+    ) -> TransactionWithSlot {
+        let mut pooled_tx = self.transaction_pool.acquire();
+        pooled_tx.reset_from_data(transaction, slot, recv_us);
+        pooled_tx.into_transaction_with_slot()
+    }
+}
+
+impl Default for ShredPoolManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// 全局 Shred 池管理器实例
+lazy_static::lazy_static! {
+    pub static ref GLOBAL_SHRED_POOL_MANAGER: ShredPoolManager = ShredPoolManager::new();
+}
+
+/// 便捷的全局工厂函数
+pub mod factory {
+    use super::*;
+
+    /// 使用对象池创建 TransactionWithSlot（推荐用于高性能场景）
+    pub fn create_transaction_with_slot_pooled(
+        transaction: VersionedTransaction,
+        slot: u64,
+        recv_us: i64,
+    ) -> TransactionWithSlot {
+        GLOBAL_SHRED_POOL_MANAGER.create_transaction_with_slot_optimized(
+            transaction, 
+            slot, 
+            recv_us
+        )
+    }
+}

src/streaming/shred/types.rs

@@ -1,11 +1,11 @@
 use solana_sdk::transaction::VersionedTransaction;
 
 /// 携带槽位信息的交易
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct TransactionWithSlot {
     pub transaction: VersionedTransaction,
     pub slot: u64,
-    pub program_received_time_us: i64,
+    pub recv_us: i64,
 }
 
 impl TransactionWithSlot {
@@ -13,8 +13,8 @@ impl TransactionWithSlot {
     pub fn new(
         transaction: VersionedTransaction,
         slot: u64,
-        program_received_time_us: i64,
+        recv_us: i64,
     ) -> Self {
-        Self { transaction, slot, program_received_time_us }
+        Self { transaction, slot, recv_us }
     }
 }

src/streaming/shred_stream.rs

@@ -8,7 +8,8 @@ use crate::protos::shredstream::SubscribeEntriesRequest;
 use crate::streaming::common::{EventProcessor, SubscriptionHandle};
 use crate::streaming::event_parser::common::filter::EventTypeFilter;
 use crate::streaming::event_parser::{Protocol, UnifiedEvent};
-use crate::streaming::shred::TransactionWithSlot;
+use crate::streaming::event_parser::core::traits::get_high_perf_clock;
+use crate::streaming::shred::pool::factory;
 use log::error;
 use solana_entry::entry::Entry;
 
@@ -57,10 +58,10 @@ impl ShredStreamGrpc {
                         if let Ok(entries) = bincode::deserialize::<Vec<Entry>>(&msg.entries) {
                             for entry in entries {
                                 for transaction in entry.transactions {
-                                    let transaction_with_slot = TransactionWithSlot::new(
+                                    let transaction_with_slot = factory::create_transaction_with_slot_pooled(
                                         transaction.clone(),
                                         msg.slot,
-                                        chrono::Utc::now().timestamp_micros(),
+                                        get_high_perf_clock(),
                                     );
                                     // 直接处理，背压控制在 EventProcessor 内部处理
                                     if let Err(e) = event_processor_clone

src/streaming/yellowstone_grpc.rs

@@ -5,8 +5,9 @@ use crate::streaming::common::{
 use crate::streaming::event_parser::common::filter::EventTypeFilter;
 use crate::streaming::event_parser::{Protocol, UnifiedEvent};
 use crate::streaming::grpc::{
-    AccountPretty, BlockMetaPretty, EventPretty, SubscriptionManager, TransactionPretty,
+    EventPretty, SubscriptionManager,
 };
+use crate::streaming::grpc::pool::factory;
 use anyhow::anyhow;
 use chrono::Local;
 use futures::channel::mpsc;
@@ -224,7 +225,7 @@ impl YellowstoneGrpc {
                                 let created_at = msg.created_at;
                                 match msg.update_oneof {
                                     Some(UpdateOneof::Account(account)) => {
-                                        let account_pretty = AccountPretty::from(account);
+                                        let account_pretty = factory::create_account_pretty_pooled(account);
                                         log::debug!("Received account: {:?}", account_pretty);
                                         if let Err(e) = event_processor
                                             .process_grpc_event_transaction_with_metrics(
@@ -237,8 +238,7 @@ impl YellowstoneGrpc {
                                         }
                                     }
                                     Some(UpdateOneof::BlockMeta(sut)) => {
-                                        let block_meta_pretty =
-                                            BlockMetaPretty::from((sut, created_at));
+                                        let block_meta_pretty = factory::create_block_meta_pretty_pooled(sut, created_at);
                                         log::debug!("Received block meta: {:?}", block_meta_pretty);
                                         if let Err(e) = event_processor
                                             .process_grpc_event_transaction_with_metrics(
@@ -251,8 +251,7 @@ impl YellowstoneGrpc {
                                         }
                                     }
                                     Some(UpdateOneof::Transaction(sut)) => {
-                                        let transaction_pretty =
-                                            TransactionPretty::from((sut, created_at));
+                                        let transaction_pretty = factory::create_transaction_pretty_pooled(sut, created_at);
                                         log::debug!(
                                             "Received transaction: {} at slot {}",
                                             transaction_pretty.signature,

src/streaming/yellowstone_sub_system.rs

@@ -1,9 +1,6 @@
 use crate::{
     common::AnyResult,
-    streaming::{
-        grpc::{EventPretty, TransactionPretty},
-        yellowstone_grpc::YellowstoneGrpc,
-    },
+    streaming::{grpc::pool::factory, grpc::EventPretty, yellowstone_grpc::YellowstoneGrpc},
 };
 use futures::{SinkExt, StreamExt};
 use log::error;
@@ -62,13 +59,11 @@ impl YellowstoneGrpc {
                         let created_at = msg.created_at;
                         match msg.update_oneof {
                             Some(UpdateOneof::Transaction(sut)) => {
-                                let transaction_pretty = TransactionPretty::from((sut, created_at));
+                                let transaction_pretty =
+                                    factory::create_transaction_pretty_pooled(sut, created_at);
                                 let event_pretty = EventPretty::Transaction(transaction_pretty);
-                                if let Err(e) = Self::process_system_transaction(
-                                    event_pretty,
-                                    &*callback,
-                                )
-                                .await
+                                if let Err(e) =
+                                    Self::process_system_transaction(event_pretty, &*callback).await
                                 {
                                     error!("Error processing transaction: {e:?}");
                                 }
@@ -105,20 +100,22 @@ impl YellowstoneGrpc {
     {
         match event_pretty {
             EventPretty::Transaction(transaction_pretty) => {
-                let trade_raw: TransactionWithStatusMeta = transaction_pretty.tx;
-                let meta = trade_raw.get_status_meta();
-
-                if meta.is_none() {
-                    return Ok(());
+                let tx = yellowstone_grpc_proto::convert_from::create_tx_with_meta(
+                    transaction_pretty.grpc_tx,
+                );
+                if let Ok(tx) = tx {
+                    let trade_raw: TransactionWithStatusMeta = tx;
+                    let meta = trade_raw.get_status_meta();
+                    if meta.is_none() {
+                        return Ok(());
+                    }
+                    let transaction = trade_raw.get_transaction();
+                    callback(SystemEvent::NewTransfer(TransferInfo {
+                        slot: transaction_pretty.slot,
+                        signature: transaction_pretty.signature.to_string(),
+                        tx: Some(transaction),
+                    }));
                 }
-
-                let transaction = trade_raw.get_transaction();
-
-                callback(SystemEvent::NewTransfer(TransferInfo {
-                    slot: transaction_pretty.slot,
-                    signature: transaction_pretty.signature.to_string(),
-                    tx: Some(transaction),
-                }));
             }
             _ => {}
         }