Sui: On-chain validator analytics with pysui

import sys
import warnings
from typing import List, Any, Tuple
from pysui import SuiConfig, SyncClient
from pysui.sui.sui_builders.get_builders import GetLatestSuiSystemState

class SuiAnalytics:
    
    def __init__(self, rpc_url: str):
        """Initialize the Sui client."""
        self.config = SuiConfig.user_config(rpc_url=rpc_url)
        self.client = SyncClient(self.config)
        self._system_state = None
        self._validators = None
    
    def _get_system_state(self) -> Any:
        """Cached getter for system state."""
        if self._system_state is None:
            result = self.client.execute(GetLatestSuiSystemState())
            if result.is_ok():
                self._system_state = result.result_data
            else:
                raise Exception(f"Failed to fetch system state: {result.result_string}")
        return self._system_state
    
    def _get_validators(self) -> List[Any]:
        """Cached getter for validators."""
        if self._validators is None:
            state = self._get_system_state()
            self._validators = state.active_validators
        return self._validators
    
    def analyze_commission_rates(self) -> None:
        """Analyze and display validator commission rates."""
        validators = self._get_validators()
        
        # Extract commission rates
        commission_rates = []
        for validator in validators:
            rate = int(validator.commission_rate) / 100
            commission_rates.append(rate)
        
        commission_rates.sort()
        
        # Calculate statistics
        # Calculate statistics
        min_commission = min(commission_rates)
        max_commission = max(commission_rates)
        avg_commission = sum(commission_rates) / len(commission_rates)
        
        # Properly calculate median for both even and odd length lists
        mid = len(commission_rates) // 2
        median_commission = commission_rates[mid] if len(commission_rates) % 2 == 1 else (commission_rates[mid-1] + commission_rates[mid]) / 2
        
        print(f"\n📊 Validator Commission Rate Analysis")
        print("=" * 50)
        print(f"Total Validators: {len(validators)}")
        print(f"Minimum Commission: {min_commission:.2f}%")
        print(f"Maximum Commission: {max_commission:.2f}%")
        print(f"Average Commission: {avg_commission:.2f}%")
        print(f"Median Commission: {median_commission:.2f}%")
        
        # Show distribution
        ranges = [(0, 1), (1, 2), (2, 3), (3, 5), (5, 10), (10, 100)]
        print(f"\n📈 Commission Rate Distribution:")
        
        for min_rate, max_rate in ranges:
            count = sum(1 for rate in commission_rates if min_rate <= rate < max_rate)
            percentage = (count / len(commission_rates)) * 100
            print(f"  {min_rate:.0f}%-{max_rate:.0f}%: {count:>3} validators ({percentage:5.1f}%)")
    
    def show_lowest_commission_validators(self, top_n: int = 10) -> None:
        """Show validators with the lowest commission rates."""
        validators = self._get_validators()
        
        # Sort by commission rate (ascending)
        sorted_by_commission = sorted(
            validators,
            key=lambda v: int(v.commission_rate)
        )
        
        print(f"\n💰 Top {top_n} Validators by Lowest Commission Rate")
        print("=" * 65)
        print(f"{'Rank':<4} {'Validator Name':<25} {'Commission':<10} {'Stake (SUI)':<15}")
        print("=" * 65)
        
        for rank, validator in enumerate(sorted_by_commission[:top_n], 1):
            name = validator.name if validator.name else "Unknown"
            name = name[:24]
            commission = int(validator.commission_rate) / 100
            stake = int(validator.staking_pool_sui_balance) / 1e9
            
            print(f"{rank:<4} {name:<25} {commission:>8.1f}% {stake:>13,.0f}")
    
    def analyze_stake_distribution(self) -> None:
        """Analyze how stake is distributed across validators."""
        validators = self._get_validators()
        
        # Get all stakes and sort them
        stakes = [int(v.staking_pool_sui_balance) / 1e9 for v in validators]
        stakes.sort(reverse=True)
        
        total_stake = sum(stakes)
        
        print(f"\n📊 Stake Distribution Analysis")
        print("=" * 50)
        print(f"Total Network Stake: {total_stake:,.0f} SUI")
        
        # Calculate concentration metrics
        top_10_stake = sum(stakes[:10])
        top_25_stake = sum(stakes[:25])
        top_50_stake = sum(stakes[:50])
        
        print(f"\nStake Concentration:")
        print(f"  Top 10 validators: {top_10_stake:,.0f} SUI ({top_10_stake/total_stake*100:.1f}%)")
        print(f"  Top 25 validators: {top_25_stake:,.0f} SUI ({top_25_stake/total_stake*100:.1f}%)")
        print(f"  Top 50 validators: {top_50_stake:,.0f} SUI ({top_50_stake/total_stake*100:.1f}%)")
        
        # Show stake ranges
        ranges = [
            (0, 1_000_000),
            (1_000_000, 5_000_000),
            (5_000_000, 10_000_000),
            (10_000_000, 25_000_000),
            (25_000_000, 50_000_000),
            (50_000_000, float('inf'))
        ]
        
        print(f"\n📈 Validator Stake Ranges:")
        for min_stake, max_stake in ranges:
            if max_stake == float('inf'):
                count = sum(1 for stake in stakes if stake >= min_stake)
                range_label = f">{min_stake/1_000_000:.0f}M SUI"
            else:
                count = sum(1 for stake in stakes if min_stake <= stake < max_stake)
                range_label = f"{min_stake/1_000_000:.0f}M-{max_stake/1_000_000:.0f}M SUI"
            
            percentage = (count / len(stakes)) * 100
            print(f"  {range_label:<15}: {count:>3} validators ({percentage:5.1f}%)")
    
    def show_epoch_info(self) -> None:
        """Display current epoch information."""
        state = self._get_system_state()
        
        print(f"\n🕐 Current Epoch Information")
        print("=" * 40)
        print(f"Current Epoch: {state.epoch}")
        print(f"System State Version: {getattr(state, 'system_state_version', 'N/A')}")
        
        # Show validator count
        validators = self._get_validators()
        print(f"Active Validators: {len(validators)}")
        
        # Calculate total stake
        total_stake = sum(int(v.staking_pool_sui_balance) for v in validators) / 1e9
        print(f"Total Staked: {total_stake:,.0f} SUI")
    
    def find_validators_by_name(self, search_term: str) -> None:
        """Search for validators by name."""
        validators = self._get_validators()
        search_lower = search_term.lower()
        
        matches = [
            v for v in validators 
            if search_lower in (v.name or "").lower()
        ]
        
        if not matches:
            print(f"\n❌ No validators found containing '{search_term}'")
            return
        
        print(f"\n🔍 Validators matching '{search_term}' ({len(matches)} found)")
        print("=" * 70)
        print(f"{'Name':<25} {'Stake (SUI)':<15} {'Commission':<10} {'Address'}")
        print("=" * 70)
        
        for validator in matches:
            name = validator.name if validator.name else "Unknown"
            name = name[:24]
            stake = int(validator.staking_pool_sui_balance) / 1e9
            commission = int(validator.commission_rate) / 100
            address = validator.sui_address[:20] + "..."
            
            print(f"{name:<25} {stake:>13,.0f} {commission:>8.1f}% {address}")
    
    def show_validator_details(self, validator_name: str) -> None:
        """Show detailed information about a specific validator."""
        validators = self._get_validators()
        
        # Find validator by name (case insensitive)
        validator = None
        for v in validators:
            if v.name and validator_name.lower() in v.name.lower():
                validator = v
                break
        
        if not validator:
            print(f"\n❌ Validator '{validator_name}' not found")
            return
        
        print(f"\n🔍 Detailed Information for '{validator.name}'")
        print("=" * 60)
        print(f"Name: {validator.name}")
        print(f"Address: {validator.sui_address}")
        print(f"Staking Pool ID: {validator.staking_pool_id}")
        print(f"Total Stake: {int(validator.staking_pool_sui_balance) / 1e9:,.0f} SUI")
        print(f"Commission Rate: {int(validator.commission_rate) / 100:.2f}%")
        print(f"Gas Price: {validator.gas_price}")
        print(f"Voting Power: {validator.voting_power}")
        
        if validator.description:
            print(f"Description: {validator.description}")
        if validator.image_url:
            print(f"Image URL: {validator.image_url}")
        if validator.project_url:
            print(f"Project URL: {validator.project_url}")
        
        print("=" * 60)


def main():
    """Main function to run Sui analytics."""
    CHAINSTACK_RPC_URL = "CHAINSTACK_NODE_URL"
    
    print("🔬 Sui analytics")
    print("=" * 50)
    
    try:
        analytics = SuiAnalytics(CHAINSTACK_RPC_URL)
        
        # Show various analytics
        analytics.show_epoch_info()
        analytics.analyze_commission_rates()
        analytics.show_lowest_commission_validators(10)
        analytics.analyze_stake_distribution()
        
        # Interactive search
        print(f"\n{'='*50}")
        search_term = input("Enter a validator name to search for (or press Enter to skip): ").strip()
        if search_term:
            analytics.find_validators_by_name(search_term)
            
            # Show detailed info if only one match
            validators = analytics._get_validators()
            matches = [v for v in validators if search_term.lower() in (v.name or "").lower()]
            if len(matches) == 1:
                print(f"\n{'='*50}")
                show_details = input("Show detailed information for this validator? (y/n): ").lower().strip()
                if show_details in ['y', 'yes']:
                    analytics.show_validator_details(search_term)
        
    except KeyboardInterrupt:
        print("\n👋 Analysis interrupted by user")
    except Exception as e:
        print(f"❌ An error occurred: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()