Module model.parts.validators
Validator Mechanisms
Validator processes such as validator activation, staking, and uptime.
Expand source code
"""
# Validator Mechanisms
Validator processes such as validator activation, staking, and uptime.
"""
import typing
import model.constants as constants
import model.parts.utils.ethereum_spec as spec
from model.types import ETH, Gwei
def policy_staking(
params, substep, state_history, previous_state
) -> typing.Dict[str, ETH]:
"""
## Staking Policy
A policy used when driving the model with the `eth_staked_process`,
for generating phase-space analyses, e.g. simulating a set of discrete `eth_staked` values.
When the `eth_staked_process` is disabled, the model is driven using the `validator_process`,
for generating state-space analyses.
"""
# Parameters
dt = params["dt"]
eth_staked_process = params["eth_staked_process"]
# State Variables
run = previous_state["run"]
timestep = previous_state["timestep"]
eth_supply = previous_state["eth_supply"]
number_of_validators = previous_state["number_of_validators"]
average_effective_balance = previous_state["average_effective_balance"]
# If the eth_staked_process is defined
if eth_staked_process(0, 0) is not None:
# Get the ETH staked sample for the current run and timestep
eth_staked = eth_staked_process(run, timestep * dt)
# Else, calculate from the number of validators
else:
eth_staked = number_of_validators * average_effective_balance / constants.gwei
# Assert expected conditions
assert eth_staked <= eth_supply, f"ETH staked can't be more than ETH supply"
return {"eth_staked": eth_staked}
def policy_validators(params, substep, state_history, previous_state):
"""
## Validator Policy Function
Calculate the number of validators driven by the ETH staked or validator processes.
"""
# Parameters
dt = params["dt"]
eth_staked_process = params["eth_staked_process"]
validator_process = params["validator_process"]
validator_uptime_process = params["validator_uptime_process"]
# State Variables
run = previous_state["run"]
timestep = previous_state["timestep"]
number_of_validators = previous_state["number_of_validators"]
number_of_validators_in_activation_queue = previous_state[
"number_of_validators_in_activation_queue"
]
average_effective_balance = previous_state["average_effective_balance"]
# Calculate the number of validators using ETH staked
if eth_staked_process(0, 0) is not None:
eth_staked = eth_staked_process(run, timestep * dt)
number_of_validators = int(
round(eth_staked / (average_effective_balance / constants.gwei))
)
else:
new_validators_per_epoch = validator_process(run, timestep * dt)
number_of_validators_in_activation_queue += new_validators_per_epoch * dt
validator_churn_limit = (
spec.get_validator_churn_limit(params, previous_state) * dt
)
activated_validators = min(
number_of_validators_in_activation_queue, validator_churn_limit
)
number_of_validators += activated_validators
number_of_validators_in_activation_queue -= activated_validators
# Calculate the number of validators online and offline using validator uptime
validator_uptime = validator_uptime_process(run, timestep * dt)
number_of_validators_online = int(round(number_of_validators * validator_uptime))
number_of_validators_offline = number_of_validators - number_of_validators_online
# Assert expected conditions
# Assume a participation of more than 2/3 due to lack of inactivity leak mechanism
assert validator_uptime >= 2 / 3, "Validator uptime must be greater than 2/3"
assert (
number_of_validators
== number_of_validators_online + number_of_validators_offline
)
return {
"number_of_validators_in_activation_queue": number_of_validators_in_activation_queue,
"number_of_validators": number_of_validators,
"number_of_validators_online": number_of_validators_online,
"number_of_validators_offline": number_of_validators_offline,
}
def policy_average_effective_balance(
params, substep, state_history, previous_state
) -> typing.Dict[str, Gwei]:
"""
## Average Effective Balance Policy Function
Calculate the validator average effective balance.
"""
# State Variables
number_of_validators = previous_state["number_of_validators"]
# Get total active balance
total_active_balance = spec.get_total_active_balance(params, previous_state)
# Aggregate by averaging over all validators
average_effective_balance = total_active_balance / number_of_validators
return {"average_effective_balance": average_effective_balance}
Functions
def policy_average_effective_balance(params, substep, state_history, previous_state) ‑> Dict[str, float]
-
Average Effective Balance Policy Function
Calculate the validator average effective balance.
Expand source code
def policy_average_effective_balance( params, substep, state_history, previous_state ) -> typing.Dict[str, Gwei]: """ ## Average Effective Balance Policy Function Calculate the validator average effective balance. """ # State Variables number_of_validators = previous_state["number_of_validators"] # Get total active balance total_active_balance = spec.get_total_active_balance(params, previous_state) # Aggregate by averaging over all validators average_effective_balance = total_active_balance / number_of_validators return {"average_effective_balance": average_effective_balance}
def policy_staking(params, substep, state_history, previous_state) ‑> Dict[str, float]
-
Staking Policy
A policy used when driving the model with the
eth_staked_process
, for generating phase-space analyses, e.g. simulating a set of discreteeth_staked
values.When the
eth_staked_process
is disabled, the model is driven using thevalidator_process
, for generating state-space analyses.Expand source code
def policy_staking( params, substep, state_history, previous_state ) -> typing.Dict[str, ETH]: """ ## Staking Policy A policy used when driving the model with the `eth_staked_process`, for generating phase-space analyses, e.g. simulating a set of discrete `eth_staked` values. When the `eth_staked_process` is disabled, the model is driven using the `validator_process`, for generating state-space analyses. """ # Parameters dt = params["dt"] eth_staked_process = params["eth_staked_process"] # State Variables run = previous_state["run"] timestep = previous_state["timestep"] eth_supply = previous_state["eth_supply"] number_of_validators = previous_state["number_of_validators"] average_effective_balance = previous_state["average_effective_balance"] # If the eth_staked_process is defined if eth_staked_process(0, 0) is not None: # Get the ETH staked sample for the current run and timestep eth_staked = eth_staked_process(run, timestep * dt) # Else, calculate from the number of validators else: eth_staked = number_of_validators * average_effective_balance / constants.gwei # Assert expected conditions assert eth_staked <= eth_supply, f"ETH staked can't be more than ETH supply" return {"eth_staked": eth_staked}
def policy_validators(params, substep, state_history, previous_state)
-
Validator Policy Function
Calculate the number of validators driven by the ETH staked or validator processes.
Expand source code
def policy_validators(params, substep, state_history, previous_state): """ ## Validator Policy Function Calculate the number of validators driven by the ETH staked or validator processes. """ # Parameters dt = params["dt"] eth_staked_process = params["eth_staked_process"] validator_process = params["validator_process"] validator_uptime_process = params["validator_uptime_process"] # State Variables run = previous_state["run"] timestep = previous_state["timestep"] number_of_validators = previous_state["number_of_validators"] number_of_validators_in_activation_queue = previous_state[ "number_of_validators_in_activation_queue" ] average_effective_balance = previous_state["average_effective_balance"] # Calculate the number of validators using ETH staked if eth_staked_process(0, 0) is not None: eth_staked = eth_staked_process(run, timestep * dt) number_of_validators = int( round(eth_staked / (average_effective_balance / constants.gwei)) ) else: new_validators_per_epoch = validator_process(run, timestep * dt) number_of_validators_in_activation_queue += new_validators_per_epoch * dt validator_churn_limit = ( spec.get_validator_churn_limit(params, previous_state) * dt ) activated_validators = min( number_of_validators_in_activation_queue, validator_churn_limit ) number_of_validators += activated_validators number_of_validators_in_activation_queue -= activated_validators # Calculate the number of validators online and offline using validator uptime validator_uptime = validator_uptime_process(run, timestep * dt) number_of_validators_online = int(round(number_of_validators * validator_uptime)) number_of_validators_offline = number_of_validators - number_of_validators_online # Assert expected conditions # Assume a participation of more than 2/3 due to lack of inactivity leak mechanism assert validator_uptime >= 2 / 3, "Validator uptime must be greater than 2/3" assert ( number_of_validators == number_of_validators_online + number_of_validators_offline ) return { "number_of_validators_in_activation_queue": number_of_validators_in_activation_queue, "number_of_validators": number_of_validators, "number_of_validators_online": number_of_validators_online, "number_of_validators_offline": number_of_validators_offline, }