Python bindings for Solana Rust tools providing high-performance blockchain development primitives, RPC functionality, and testing infrastructure.
—
Network configuration, system variables, and blockchain state including clock information, rent parameters, epoch scheduling, and validator information. These provide essential network parameters and real-time blockchain state for building robust Solana applications.
System accounts that contain network configuration and state information accessible to all programs.
# Sysvar Account IDs - Well-known addresses for system variables
CLOCK: Final[Pubkey] = Pubkey.from_string("SysvarC1ock11111111111111111111111111111111")
RECENT_BLOCKHASHES: Final[Pubkey] = Pubkey.from_string("SysvarRecentB1ockHashes11111111111111111111")
RENT: Final[Pubkey] = Pubkey.from_string("SysvarRent111111111111111111111111111111111")
REWARDS: Final[Pubkey] = Pubkey.from_string("SysvarRewards111111111111111111111111111111")
STAKE_HISTORY: Final[Pubkey] = Pubkey.from_string("SysvarStakeHistory1111111111111111111111111")
EPOCH_SCHEDULE: Final[Pubkey] = Pubkey.from_string("SysvarEpochSchedu1e111111111111111111111111111")
INSTRUCTIONS: Final[Pubkey] = Pubkey.from_string("Sysvar1nstructions1111111111111111111111111")
SLOT_HASHES: Final[Pubkey] = Pubkey.from_string("SysvarS1otHashes111111111111111111111111111")Network time and slot information for transaction timing and epoch tracking.
class Clock:
"""
Network time information including slot, epoch, and timestamp data.
"""
def __init__(
self,
slot: int,
epoch_start_timestamp: Optional[int],
epoch: int,
leader_schedule_epoch: int,
unix_timestamp: int
):
"""
Create clock information.
Parameters:
- slot: int, current slot number
- epoch_start_timestamp: Optional[int], Unix timestamp when current epoch started
- epoch: int, current epoch number
- leader_schedule_epoch: int, epoch for which leader schedule is valid
- unix_timestamp: int, estimated current Unix timestamp
"""
@classmethod
def deserialize(cls, data: bytes) -> 'Clock':
"""
Deserialize clock from sysvar account data.
Parameters:
- data: bytes, clock sysvar account data
Returns:
Clock object
"""
def serialize(self) -> bytes:
"""
Serialize clock to bytes.
Returns:
bytes, serialized clock data
"""
@property
def slot(self) -> int:
"""Current slot number."""
@property
def epoch_start_timestamp(self) -> Optional[int]:
"""Unix timestamp when current epoch started."""
@property
def epoch(self) -> int:
"""Current epoch number."""
@property
def leader_schedule_epoch(self) -> int:
"""Epoch for which leader schedule is valid."""
@property
def unix_timestamp(self) -> int:
"""Estimated current Unix timestamp."""
def slots_since_epoch_start(self) -> int:
"""
Calculate slots elapsed in current epoch.
Returns:
int, number of slots since epoch started
"""
def time_since_epoch_start(self) -> Optional[int]:
"""
Calculate time elapsed in current epoch.
Returns:
Optional[int], seconds since epoch started (None if no epoch start timestamp)
"""Fundamental timing parameters that define network behavior and performance characteristics.
# Default timing configuration for mainnet
DEFAULT_DEV_SLOTS_PER_EPOCH: Final[int] = 8192 # Development/testnet epoch length
DEFAULT_SLOTS_PER_EPOCH: Final[int] = 432000 # Production epoch length (~2 days)
DEFAULT_MS_PER_SLOT: Final[int] = 400 # Target slot duration (400ms)
DEFAULT_S_PER_SLOT: Final[float] = 0.4 # Target slot duration (0.4s)
DEFAULT_TICKS_PER_SLOT: Final[int] = 64 # Ticks per slot
DEFAULT_TICKS_PER_SECOND: Final[int] = 160 # Target tick rate
DEFAULT_HASHES_PER_SECOND: Final[int] = 1000000 # Hash rate target
DEFAULT_HASHES_PER_TICK: Final[int] = 6250 # Hashes per tick
# Epoch and slot configuration
GENESIS_EPOCH: Final[int] = 0 # First epoch number
INITIAL_RENT_EPOCH: Final[int] = 0 # Initial rent collection epoch
# Transaction timing limits
MAX_HASH_AGE_IN_SECONDS: Final[int] = 120 # Maximum blockhash age (2 minutes)
MAX_PROCESSING_AGE: Final[int] = 150 # Maximum transaction processing age
MAX_RECENT_BLOCKHASHES: Final[int] = 300 # Maximum tracked recent blockhashes
# Leader scheduling
NUM_CONSECUTIVE_LEADER_SLOTS: Final[int] = 4 # Consecutive slots per leader
FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET: Final[int] = 2 # Transaction forwarding offset
HOLD_TRANSACTIONS_SLOT_OFFSET: Final[int] = 20 # Transaction hold duration
# GPU-specific timing
MAX_TRANSACTION_FORWARDING_DELAY: Final[int] = 6 # Max forwarding delay (slots)
MAX_TRANSACTION_FORWARDING_DELAY_GPU: Final[int] = 2 # Max GPU forwarding delay
# Time constants
MS_PER_TICK: Final[float] = 6.25 # Milliseconds per tick
SECONDS_PER_DAY: Final[int] = 86400 # Seconds per day
TICKS_PER_DAY: Final[int] = 13824000 # Ticks per day (86400 / 0.0625 * 1000)Account rent parameters and exemption thresholds for storage cost calculation.
class Rent:
"""
Rent calculation configuration and exemption parameters.
"""
def __init__(
self,
lamports_per_byte_year: int,
exemption_threshold: float,
burn_percent: int
):
"""
Create rent configuration.
Parameters:
- lamports_per_byte_year: int, cost per byte per year in lamports
- exemption_threshold: float, years worth of rent for exemption
- burn_percent: int, percentage of collected rent to burn
"""
@classmethod
def default() -> 'Rent':
"""
Create default rent configuration.
Returns:
Rent with mainnet default parameters
"""
@classmethod
def deserialize(cls, data: bytes) -> 'Rent':
"""
Deserialize rent from sysvar account data.
Parameters:
- data: bytes, rent sysvar account data
Returns:
Rent object
"""
def serialize(self) -> bytes:
"""
Serialize rent to bytes.
Returns:
bytes, serialized rent data
"""
@property
def lamports_per_byte_year(self) -> int:
"""Cost per byte per year in lamports."""
@property
def exemption_threshold(self) -> float:
"""Years worth of rent required for exemption."""
@property
def burn_percent(self) -> int:
"""Percentage of collected rent to burn (0-100)."""
def minimum_balance(self, data_len: int) -> int:
"""
Calculate minimum lamports for rent exemption.
Parameters:
- data_len: int, account data size in bytes
Returns:
int, minimum lamports needed for rent exemption
"""
def due(self, lamports: int, data_len: int, years_elapsed: float) -> int:
"""
Calculate rent due for account.
Parameters:
- lamports: int, current account balance
- data_len: int, account data size
- years_elapsed: float, time elapsed since last rent payment
Returns:
int, rent due in lamports (0 if rent exempt)
"""
def is_exempt(self, lamports: int, data_len: int) -> bool:
"""
Check if account is rent exempt.
Parameters:
- lamports: int, account balance
- data_len: int, account data size
Returns:
bool, True if account is rent exempt
"""
# Rent configuration constants
ACCOUNT_STORAGE_OVERHEAD: Final[int] = 128 # Additional bytes per account
DEFAULT_LAMPORTS_PER_BYTE_YEAR: Final[int] = 3480 # Default storage cost
DEFAULT_EXEMPTION_THRESHOLD: Final[float] = 2.0 # Default exemption threshold (2 years)
DEFAULT_BURN_PERCENT: Final[int] = 50 # Default burn percentageEpoch timing, slot allocation, and validator scheduling parameters.
class EpochInfo:
"""
Current epoch information including progress and timing.
"""
def __init__(
self,
epoch: int,
slot_index: int,
slots_in_epoch: int,
absolute_slot: int,
block_height: int,
transaction_count: Optional[int]
):
"""
Create epoch information.
Parameters:
- epoch: int, current epoch number
- slot_index: int, slot index within current epoch
- slots_in_epoch: int, total slots in current epoch
- absolute_slot: int, absolute slot number since genesis
- block_height: int, current block height
- transaction_count: Optional[int], total transaction count
"""
@property
def epoch(self) -> int:
"""Current epoch number."""
@property
def slot_index(self) -> int:
"""Slot index within current epoch."""
@property
def slots_in_epoch(self) -> int:
"""Total slots in current epoch."""
@property
def absolute_slot(self) -> int:
"""Absolute slot number since genesis."""
@property
def block_height(self) -> int:
"""Current block height."""
@property
def transaction_count(self) -> Optional[int]:
"""Total transaction count (if available)."""
def progress(self) -> float:
"""
Calculate epoch completion percentage.
Returns:
float, completion percentage (0.0 to 1.0)
"""
def slots_remaining(self) -> int:
"""
Calculate slots remaining in epoch.
Returns:
int, number of slots until epoch end
"""
class EpochSchedule:
"""
Epoch scheduling configuration and timing parameters.
"""
def __init__(
self,
slots_per_epoch: int,
leader_schedule_slot_offset: int,
warmup: bool,
first_normal_epoch: int,
first_normal_slot: int
):
"""
Create epoch schedule configuration.
Parameters:
- slots_per_epoch: int, slots per epoch after warmup
- leader_schedule_slot_offset: int, offset for leader schedule calculation
- warmup: bool, whether network is in warmup period
- first_normal_epoch: int, first epoch with normal slot count
- first_normal_slot: int, first slot of first normal epoch
"""
@classmethod
def default() -> 'EpochSchedule':
"""
Create default epoch schedule.
Returns:
EpochSchedule with mainnet parameters
"""
@classmethod
def deserialize(cls, data: bytes) -> 'EpochSchedule':
"""
Deserialize epoch schedule from sysvar account data.
Parameters:
- data: bytes, epoch schedule sysvar data
Returns:
EpochSchedule object
"""
def serialize(self) -> bytes:
"""
Serialize epoch schedule to bytes.
Returns:
bytes, serialized schedule data
"""
@property
def slots_per_epoch(self) -> int:
"""Slots per epoch after warmup."""
@property
def leader_schedule_slot_offset(self) -> int:
"""Offset for leader schedule calculation."""
@property
def warmup(self) -> bool:
"""Whether network is in warmup period."""
@property
def first_normal_epoch(self) -> int:
"""First epoch with normal slot count."""
@property
def first_normal_slot(self) -> int:
"""First slot of first normal epoch."""
def get_epoch(self, slot: int) -> int:
"""
Calculate epoch number for given slot.
Parameters:
- slot: int, slot number
Returns:
int, epoch containing the slot
"""
def get_epoch_and_slot_index(self, slot: int) -> tuple[int, int]:
"""
Calculate epoch and slot index for given slot.
Parameters:
- slot: int, slot number
Returns:
tuple[int, int], (epoch, slot_index_in_epoch)
"""
def get_first_slot_in_epoch(self, epoch: int) -> int:
"""
Calculate first slot of given epoch.
Parameters:
- epoch: int, epoch number
Returns:
int, first slot in epoch
"""
def get_last_slot_in_epoch(self, epoch: int) -> int:
"""
Calculate last slot of given epoch.
Parameters:
- epoch: int, epoch number
Returns:
int, last slot in epoch
"""
def get_slots_in_epoch(self, epoch: int) -> int:
"""
Calculate number of slots in given epoch.
Parameters:
- epoch: int, epoch number
Returns:
int, slots in epoch
"""Historical stake information and validator performance tracking.
class StakeHistory:
"""
Historical stake activation information across epochs.
"""
def __init__(self, entries: List['StakeHistoryEntry']):
"""
Create stake history with epoch entries.
Parameters:
- entries: List[StakeHistoryEntry], historical stake data
"""
@classmethod
def deserialize(cls, data: bytes) -> 'StakeHistory':
"""
Deserialize stake history from sysvar account data.
Parameters:
- data: bytes, stake history sysvar data
Returns:
StakeHistory object
"""
def serialize(self) -> bytes:
"""
Serialize stake history to bytes.
Returns:
bytes, serialized stake history
"""
@property
def entries(self) -> List['StakeHistoryEntry']:
"""Historical stake entries by epoch."""
def get_stake_history_entry(self, epoch: int) -> Optional['StakeHistoryEntry']:
"""
Get stake information for specific epoch.
Parameters:
- epoch: int, epoch number to query
Returns:
Optional[StakeHistoryEntry], stake info for epoch (None if not found)
"""
class StakeHistoryEntry:
"""
Stake information for a single epoch.
"""
def __init__(self, effective: int, activating: int, deactivating: int):
"""
Create stake history entry.
Parameters:
- effective: int, effective stake amount in lamports
- activating: int, stake being activated in lamports
- deactivating: int, stake being deactivated in lamports
"""
@property
def effective(self) -> int:
"""Effective stake amount in lamports."""
@property
def activating(self) -> int:
"""Stake being activated in lamports."""
@property
def deactivating(self) -> int:
"""Stake being deactivated in lamports."""
def total_stake(self) -> int:
"""
Calculate total stake (effective + activating).
Returns:
int, total stake amount
"""
class SlotHistory:
"""
Historical slot information and confirmation bitmap.
"""
def __init__(self, bits: bytes, next_slot: int):
"""
Create slot history with confirmation bitmap.
Parameters:
- bits: bytes, bitmap of confirmed slots
- next_slot: int, next slot to be recorded
"""
@classmethod
def deserialize(cls, data: bytes) -> 'SlotHistory':
"""
Deserialize slot history from sysvar account data.
Returns:
SlotHistory object
"""
def serialize(self) -> bytes:
"""Serialize slot history to bytes."""
@property
def bits(self) -> bytes:
"""Bitmap of confirmed slots."""
@property
def next_slot(self) -> int:
"""Next slot to be recorded."""
def check_slot(self, slot: int) -> 'SlotHistoryCheck':
"""
Check if slot is in history and confirmed.
Parameters:
- slot: int, slot number to check
Returns:
SlotHistoryCheck, validation result
"""
class SlotHistoryCheck:
"""Result of slot history validation."""
Found: 'SlotHistoryCheck' # Slot found and confirmed
NotFound: 'SlotHistoryCheck' # Slot not in history
TooOld: 'SlotHistoryCheck' # Slot too old (outside history window)
TooNew: 'SlotHistoryCheck' # Slot too new (future slot)Epoch reward distribution and inflation parameters.
class EpochRewards:
"""
Epoch reward distribution information.
"""
def __init__(
self,
distribution_starting_block_height: int,
num_partitions: int,
parent_blockhash: Hash,
total_reward: int,
distributed_rewards: int,
active: bool
):
"""
Create epoch rewards information.
Parameters:
- distribution_starting_block_height: int, block height when distribution started
- num_partitions: int, number of reward distribution partitions
- parent_blockhash: Hash, parent blockhash for reward calculation
- total_reward: int, total rewards for epoch in lamports
- distributed_rewards: int, rewards distributed so far in lamports
- active: bool, whether reward distribution is active
"""
@property
def distribution_starting_block_height(self) -> int:
"""Block height when distribution started."""
@property
def num_partitions(self) -> int:
"""Number of reward distribution partitions."""
@property
def parent_blockhash(self) -> Hash:
"""Parent blockhash for reward calculation."""
@property
def total_reward(self) -> int:
"""Total rewards for epoch in lamports."""
@property
def distributed_rewards(self) -> int:
"""Rewards distributed so far in lamports."""
@property
def active(self) -> bool:
"""Whether reward distribution is active."""
def remaining_rewards(self) -> int:
"""
Calculate remaining rewards to distribute.
Returns:
int, undistributed rewards in lamports
"""
def distribution_progress(self) -> float:
"""
Calculate reward distribution progress.
Returns:
float, distribution completion (0.0 to 1.0)
"""from solders.clock import Clock
from solders.rent import Rent
from solders.epoch_schedule import EpochSchedule
from solders.sysvar import CLOCK, RENT, EPOCH_SCHEDULE
from solders.rpc.requests import GetAccountInfo
# Read clock sysvar
clock_request = GetAccountInfo(CLOCK)
# After RPC call, deserialize the account data:
# clock = Clock.deserialize(account.data)
# Read rent sysvar
rent_request = GetAccountInfo(RENT)
# rent = Rent.deserialize(account.data)
# Read epoch schedule
schedule_request = GetAccountInfo(EPOCH_SCHEDULE)
# epoch_schedule = EpochSchedule.deserialize(account.data)# Assuming clock data was retrieved from RPC
clock_data = bytes(40) # Clock sysvar data from RPC response
clock = Clock.deserialize(clock_data)
print(f"Current slot: {clock.slot}")
print(f"Current epoch: {clock.epoch}")
print(f"Unix timestamp: {clock.unix_timestamp}")
# Calculate elapsed time in epoch
slots_elapsed = clock.slots_since_epoch_start()
print(f"Slots since epoch start: {slots_elapsed}")
# Estimate time since epoch start
if clock.epoch_start_timestamp:
time_elapsed = clock.time_since_epoch_start()
print(f"Seconds since epoch start: {time_elapsed}")
# Calculate slot timing
slot_duration_ms = DEFAULT_MS_PER_SLOT
estimated_next_slot_time = clock.unix_timestamp + (slot_duration_ms / 1000)
print(f"Estimated next slot at: {estimated_next_slot_time}")# Example rent calculations
rent = Rent.default()
# Calculate rent exemption for different account types
token_account_size = 165
minimum_for_token = rent.minimum_balance(token_account_size)
print(f"Minimum for token account: {minimum_for_token} lamports ({minimum_for_token / 1e9:.9f} SOL)")
mint_account_size = 82
minimum_for_mint = rent.minimum_balance(mint_account_size)
print(f"Minimum for mint account: {minimum_for_mint} lamports ({minimum_for_mint / 1e9:.9f} SOL)")
# Check if account is rent exempt
account_balance = 2500000 # 0.0025 SOL
account_size = 100
if rent.is_exempt(account_balance, account_size):
print("Account is rent exempt")
else:
needed = rent.minimum_balance(account_size) - account_balance
print(f"Need {needed} more lamports for rent exemption")
# Calculate rent due (for non-exempt accounts)
years_elapsed = 0.1 # ~36 days
rent_due = rent.due(account_balance, account_size, years_elapsed)
print(f"Rent due: {rent_due} lamports")# Working with epoch schedule
epoch_schedule = EpochSchedule.default()
current_slot = 100000000
current_epoch = epoch_schedule.get_epoch(current_slot)
epoch_info = epoch_schedule.get_epoch_and_slot_index(current_slot)
print(f"Slot {current_slot} is in epoch {current_epoch}")
print(f"Epoch: {epoch_info[0]}, Slot index: {epoch_info[1]}")
# Calculate epoch boundaries
first_slot = epoch_schedule.get_first_slot_in_epoch(current_epoch)
last_slot = epoch_schedule.get_last_slot_in_epoch(current_epoch)
slots_in_epoch = epoch_schedule.get_slots_in_epoch(current_epoch)
print(f"Epoch {current_epoch}: slots {first_slot} to {last_slot} ({slots_in_epoch} total)")
# Calculate progress through epoch
slot_index = current_slot - first_slot
progress = slot_index / slots_in_epoch
print(f"Epoch progress: {progress:.2%}")
# Estimate time remaining in epoch
slots_remaining = last_slot - current_slot
time_remaining_ms = slots_remaining * DEFAULT_MS_PER_SLOT
hours_remaining = (time_remaining_ms / 1000) / 3600
print(f"Estimated time remaining in epoch: {hours_remaining:.1f} hours")# Working with stake history (example)
# stake_history = StakeHistory.deserialize(stake_history_sysvar_data)
def analyze_stake_trend(stake_history: StakeHistory, epochs_back: int = 10):
"""Analyze stake trend over recent epochs."""
current_epoch = 500 # Would get from clock or epoch info
stake_values = []
for i in range(epochs_back):
epoch = current_epoch - i
entry = stake_history.get_stake_history_entry(epoch)
if entry:
total_stake = entry.total_stake()
stake_values.append((epoch, total_stake))
if len(stake_values) >= 2:
recent = stake_values[0][1]
older = stake_values[-1][1]
change = ((recent - older) / older) * 100
print(f"Stake change over {epochs_back} epochs: {change:+.2f}%")
return stake_values
# Check slot confirmation
# slot_history = SlotHistory.deserialize(slot_history_sysvar_data)
def check_slot_confirmation(slot_history: SlotHistory, slot: int):
"""Check if a slot was confirmed."""
result = slot_history.check_slot(slot)
if result == SlotHistoryCheck.Found:
print(f"Slot {slot} was confirmed")
elif result == SlotHistoryCheck.NotFound:
print(f"Slot {slot} was not confirmed")
elif result == SlotHistoryCheck.TooOld:
print(f"Slot {slot} is too old (outside history window)")
elif result == SlotHistoryCheck.TooNew:
print(f"Slot {slot} is too new (future slot)")# Calculate network timing metrics
def calculate_network_metrics(current_slot: int, target_slot: int):
"""Calculate timing for reaching target slot."""
slots_to_wait = target_slot - current_slot
if slots_to_wait <= 0:
print("Target slot already passed")
return
# Time estimates
ms_to_wait = slots_to_wait * DEFAULT_MS_PER_SLOT
seconds_to_wait = ms_to_wait / 1000
minutes_to_wait = seconds_to_wait / 60
print(f"Slots to wait: {slots_to_wait}")
print(f"Estimated time: {minutes_to_wait:.1f} minutes")
# Account for network variability
min_time = seconds_to_wait * 0.8 # Slots can be faster
max_time = seconds_to_wait * 1.2 # Slots can be slower
print(f"Time range: {min_time/60:.1f} - {max_time/60:.1f} minutes")
# Transaction timing validation
def validate_transaction_timing(blockhash_age_slots: int) -> bool:
"""Validate if blockhash is still valid for transactions."""
max_age_slots = MAX_HASH_AGE_IN_SECONDS / DEFAULT_S_PER_SLOT
if blockhash_age_slots > max_age_slots:
print(f"Blockhash too old: {blockhash_age_slots} slots (max {max_age_slots})")
return False
slots_remaining = max_age_slots - blockhash_age_slots
time_remaining = slots_remaining * DEFAULT_S_PER_SLOT
print(f"Blockhash valid for {slots_remaining:.0f} more slots ({time_remaining:.0f}s)")
return True
# Usage
current_slot = 100000000
target_slot = 100000100
calculate_network_metrics(current_slot, target_slot)
# Check blockhash validity
blockhash_slot = 99999900 # Slot when blockhash was created
age_in_slots = current_slot - blockhash_slot
is_valid = validate_transaction_timing(age_in_slots)# Working with epoch rewards (example)
def track_reward_distribution(epoch_rewards: EpochRewards):
"""Track epoch reward distribution progress."""
print(f"Total rewards: {epoch_rewards.total_reward / 1e9:.6f} SOL")
print(f"Distributed: {epoch_rewards.distributed_rewards / 1e9:.6f} SOL")
remaining = epoch_rewards.remaining_rewards()
print(f"Remaining: {remaining / 1e9:.6f} SOL")
progress = epoch_rewards.distribution_progress()
print(f"Distribution progress: {progress:.1%}")
if epoch_rewards.active:
print("Distribution is currently active")
# Estimate completion time based on distribution rate
if progress > 0:
# This would require historical data to calculate rate
print("Estimating completion time...")
else:
print("Distribution completed or not started")# Network-specific parameters (examples for different networks)
class NetworkParams:
# Mainnet parameters
MAINNET_GENESIS_HASH = "5eykt4UsFv8P8NJdTREpY1vzqKqZKvdpKuc147dw2N9d"
# Devnet parameters
DEVNET_SLOTS_PER_EPOCH = DEFAULT_DEV_SLOTS_PER_EPOCH
# Testnet parameters
TESTNET_SLOTS_PER_EPOCH = DEFAULT_SLOTS_PER_EPOCH# Network performance targets
TARGET_TRANSACTIONS_PER_SECOND = 65000 # Peak TPS target
TARGET_CONFIRMATION_TIME_MS = 400 # Target confirmation time
TARGET_FINALIZATION_TIME_MS = 12800 # Target finalization time (32 slots)
# Practical limits
MAX_TRANSACTIONS_PER_BLOCK = 512 # Current block limit
AVERAGE_TRANSACTION_SIZE = 200 # Average transaction size in bytes