import asyncio
import sys
from pathlib import Path
from typing import Any, List
from pysui import PysuiConfiguration
from pysui.sui.sui_common.config.confgroup import NetworkType
from pysui.sui.sui_grpc.pgrpc_clients import GrpcProtocolClient
import pysui.sui.sui_grpc.pgrpc_requests as rn
CHAINSTACK_GRPC_URL = "https://sui-mainnet.core.chainstack.com:443"
CHAINSTACK_X_TOKEN = "YOUR_X_TOKEN"
CONFIG_DIR = Path("~/.pysui-chainstack").expanduser()
def load_config() -> PysuiConfiguration:
"""Load the pysui configuration, creating a Chainstack profile on first run."""
try:
return PysuiConfiguration(from_cfg_path=str(CONFIG_DIR), group_name="chainstack")
except ValueError:
cfg = PysuiConfiguration.initialize_config(
in_folder=CONFIG_DIR,
init_groups=[{"name": "chainstack", "make_active": True}],
)
cfg.new_profile(
profile_name="mainnet",
url=CHAINSTACK_GRPC_URL,
network_type=NetworkType.PRODUCTION,
make_active=True,
in_group="chainstack",
)
return cfg
class SuiAnalytics:
def __init__(self, cfg: PysuiConfiguration, x_token: str):
self.cfg = cfg
self.x_token = x_token
self._system_state = None
async def load(self) -> None:
"""Fetch the current system state, which holds the active validator set."""
client = GrpcProtocolClient(pysui_config=self.cfg, default_header={"x-token": self.x_token})
result = await client.execute_grpc_request(request=rn.GetLatestSuiSystemState())
await client.close()
if not result.is_ok():
raise Exception(f"Failed to fetch system state: {result.result_string}")
self._system_state = result.result_data
def _get_validators(self) -> List[Any]:
return self._system_state.validators.active_validators
def analyze_commission_rates(self) -> None:
"""Analyze and display validator commission rates."""
validators = self._get_validators()
# Commission rates are in basis points
commission_rates = sorted(int(v.commission_rate) / 100 for v in validators)
min_commission = min(commission_rates)
max_commission = max(commission_rates)
avg_commission = sum(commission_rates) / len(commission_rates)
# 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}%")
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()
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 or "Unknown")[: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()
stakes = sorted((int(v.staking_pool.sui_balance) / 1e9 for v in validators), reverse=True)
total_stake = sum(stakes)
print(f"\n📊 Stake Distribution Analysis")
print("=" * 50)
print(f"Total Network Stake: {total_stake:,.0f} SUI")
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}%)")
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._system_state
validators = self._get_validators()
total_stake = sum(int(v.staking_pool.sui_balance) for v in validators) / 1e9
print(f"\n🕐 Current Epoch Information")
print("=" * 40)
print(f"Current Epoch: {state.epoch}")
print(f"Protocol Version: {state.protocol_version}")
print(f"Active Validators: {len(validators)}")
print(f"Total Staked: {total_stake:,.0f} SUI")
def find_validators_by_name(self, search_term: str) -> List[Any]:
"""Search for validators by name."""
search_lower = search_term.lower()
matches = [v for v in self._get_validators() if search_lower in (v.name or "").lower()]
if not matches:
print(f"\n❌ No validators found containing '{search_term}'")
return matches
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 or "Unknown")[:24]
stake = int(validator.staking_pool.sui_balance) / 1e9
commission = int(validator.commission_rate) / 100
print(f"{name:<25} {stake:>13,.0f} {commission:>8.1f}% {validator.address}")
return matches
def show_validator_details(self, validator: Any) -> None:
"""Show detailed information about a validator."""
print(f"\n🔍 Detailed Information for '{validator.name}'")
print("=" * 60)
print(f"Name: {validator.name}")
print(f"Address: {validator.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."""
print("🔬 Sui analytics")
print("=" * 50)
try:
analytics = SuiAnalytics(load_config(), CHAINSTACK_X_TOKEN)
asyncio.run(analytics.load())
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:
matches = analytics.find_validators_by_name(search_term)
# Offer details if there is exactly one match
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(matches[0])
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()
