Module model.parts.ethereum_system
Ethereum System
General Ethereum mechanisms, such as managing the system upgrade process, the EIP-1559 transaction pricing mechanism, and updating the ETH price and ETH supply.
Expand source code
"""
# Ethereum System
General Ethereum mechanisms, such as managing the system upgrade process,
the EIP-1559 transaction pricing mechanism, and updating the ETH price and ETH supply.
"""
import typing
import datetime
from model import constants as constants
from model.types import ETH, USD_per_ETH, Gwei, Stage
def policy_upgrade_stages(params, substep, state_history, previous_state):
"""
## Upgrade Stages Policy
Transitions the model from one stage in the Ethereum network
upgrade process to the next at different milestones.
This is essentially a finite-state machine: https://en.wikipedia.org/wiki/Finite-state_machine
"""
# Parameters
dt = params["dt"]
stage: Stage = params["stage"]
date_start = params["date_start"]
date_eip1559 = params["date_eip1559"]
date_pos = params["date_pos"]
# State Variables
current_stage = previous_state["stage"]
timestep = previous_state["timestep"]
# Calculate current timestamp from timestep
timestamp = date_start + datetime.timedelta(
days=(timestep * dt / constants.epochs_per_day)
)
# Initialize stage State Variable at start of simulation
if current_stage is None:
current_stage = stage
else:
# Convert Stage enum value (int) to Stage enum
current_stage = Stage(current_stage)
# Stage finite-state machine
if stage == Stage.ALL:
# If Stage ALL selected, transition through all stages
# at different timestamps
if (
current_stage in [Stage.ALL, Stage.BEACON_CHAIN]
and timestamp < date_eip1559
):
current_stage = Stage.BEACON_CHAIN
elif (
current_stage in [Stage.BEACON_CHAIN, Stage.EIP1559]
and timestamp < date_pos
):
current_stage = Stage.EIP1559
else:
current_stage = Stage.PROOF_OF_STAKE
elif stage == Stage.BEACON_CHAIN:
# If Stage BEACON_CHAIN selected, only execute single stage
current_stage = Stage.BEACON_CHAIN
elif stage == Stage.EIP1559:
# If Stage EIP1559 selected, only execute single stage
current_stage = Stage.EIP1559
elif stage == Stage.PROOF_OF_STAKE:
# If Stage PROOF_OF_STAKE selected, only execute single stage
current_stage = Stage.PROOF_OF_STAKE
else:
# Else, raise exception if invalid Stage
raise Exception("Invalid Stage selected")
return {
"stage": current_stage.value,
"timestamp": timestamp,
}
def policy_network_issuance(
params, substep, state_history, previous_state
) -> typing.Dict[str, ETH]:
"""
## Network Issuance Policy Function
Calculate the total network issuance and issuance from Proof of Work block rewards.
"""
# Parameters
dt = params["dt"]
daily_pow_issuance = params["daily_pow_issuance"]
# State Variables
stage = previous_state["stage"]
amount_slashed = previous_state["amount_slashed"]
total_base_fee = previous_state["total_base_fee"]
total_priority_fee_to_validators = previous_state[
"total_priority_fee_to_validators"
]
total_online_validator_rewards = previous_state["total_online_validator_rewards"]
# Calculate network issuance in ETH
network_issuance = (
# Remove priority fee to validators which is not issuance (ETH transferred rather than minted)
(total_online_validator_rewards - total_priority_fee_to_validators)
- amount_slashed
- total_base_fee
) / constants.gwei
# Calculate Proof of Work issuance
pow_issuance = (
daily_pow_issuance / constants.epochs_per_day
if Stage(stage) in [Stage.BEACON_CHAIN, Stage.EIP1559]
else 0
)
network_issuance += pow_issuance * dt
return {
"network_issuance": network_issuance,
"pow_issuance": pow_issuance,
}
def policy_eip1559_transaction_pricing(
params, substep, state_history, previous_state
) -> typing.Dict[str, Gwei]:
"""
## EIP1559 Transaction Pricing Mechanism
A transaction pricing mechanism that includes fixed-per-block network fee
that is burned and dynamically expands/contracts block sizes to deal with transient congestion.
See:
* https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1559.md
* https://eips.ethereum.org/EIPS/eip-1559
"""
stage = Stage(previous_state["stage"])
if stage not in [Stage.EIP1559, Stage.PROOF_OF_STAKE]:
return {
"base_fee_per_gas": 0,
"total_base_fee": 0,
"total_priority_fee_to_miners": 0,
"total_priority_fee_to_validators": 0,
}
# Parameters
dt = params["dt"]
gas_target_process = params["gas_target_process"] # Gas
ELASTICITY_MULTIPLIER = params["ELASTICITY_MULTIPLIER"]
base_fee_process = params["base_fee_process"]
priority_fee_process = params["priority_fee_process"]
# State Variables
run = previous_state["run"]
timestep = previous_state["timestep"]
# Get samples for current run and timestep from base fee, priority fee, and transaction processes
base_fee_per_gas = base_fee_process(run, timestep * dt) # Gwei per Gas
gas_target = gas_target_process(run, timestep * dt) # Gas
# Ensure basefee changes by no more than 1 / BASE_FEE_MAX_CHANGE_DENOMINATOR %
_BASE_FEE_MAX_CHANGE_DENOMINATOR = params["BASE_FEE_MAX_CHANGE_DENOMINATOR"]
# assert (
# abs(basefee - previous_basefee) / previous_basefee
# <= constants.slots_per_epoch / BASE_FEE_MAX_CHANGE_DENOMINATOR
# if timestep > 1
# else True
# ), "basefee changed by more than 1 / BASE_FEE_MAX_CHANGE_DENOMINATOR %"
avg_priority_fee_per_gas = priority_fee_process(run, timestep * dt) # Gwei per Gas
if stage in [Stage.EIP1559]:
gas_used = constants.pow_blocks_per_epoch * gas_target # Gas
else: # stage is Stage.PROOF_OF_STAKE
gas_used = constants.slots_per_epoch * gas_target # Gas
# Calculate total base fee, and priority fee to validators
total_base_fee = gas_used * base_fee_per_gas # Gwei
total_priority_fee = gas_used * avg_priority_fee_per_gas # Gwei
if stage in [Stage.PROOF_OF_STAKE]:
total_priority_fee_to_miners = 0
total_priority_fee_to_validators = total_priority_fee
else:
total_priority_fee_to_miners = total_priority_fee
total_priority_fee_to_validators = 0
# Check if the block used too much gas
assert (
gas_used <= gas_target * ELASTICITY_MULTIPLIER * constants.slots_per_epoch
), "invalid block: too much gas used"
return {
"base_fee_per_gas": base_fee_per_gas,
"total_base_fee": total_base_fee * dt,
"total_priority_fee_to_miners": total_priority_fee_to_miners * dt,
"total_priority_fee_to_validators": total_priority_fee_to_validators * dt,
}
def update_eth_price(
params, substep, state_history, previous_state, policy_input
) -> typing.Tuple[str, USD_per_ETH]:
"""
## ETH Price State Update Function
Update the ETH price from the `eth_price_process`.
"""
# Parameters
dt = params["dt"]
eth_price_process = params["eth_price_process"]
# State Variables
run = previous_state["run"]
timestep = previous_state["timestep"]
# Get the ETH price sample for the current run and timestep
eth_price_sample = eth_price_process(run, timestep * dt)
return "eth_price", eth_price_sample
def update_eth_supply(
params, substep, state_history, previous_state, policy_input
) -> typing.Tuple[str, ETH]:
"""
## ETH Supply State Update Function
Update the ETH supply from the Network Issuance Policy Function.
"""
# Policy Inputs
network_issuance = policy_input["network_issuance"]
# State variables
eth_supply = previous_state["eth_supply"]
return "eth_supply", eth_supply + network_issuance
Functions
def policy_eip1559_transaction_pricing(params, substep, state_history, previous_state) ‑> Dict[str, float]
-
EIP1559 Transaction Pricing Mechanism
A transaction pricing mechanism that includes fixed-per-block network fee that is burned and dynamically expands/contracts block sizes to deal with transient congestion.
See: * https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1559.md * https://eips.ethereum.org/EIPS/eip-1559
Expand source code
def policy_eip1559_transaction_pricing( params, substep, state_history, previous_state ) -> typing.Dict[str, Gwei]: """ ## EIP1559 Transaction Pricing Mechanism A transaction pricing mechanism that includes fixed-per-block network fee that is burned and dynamically expands/contracts block sizes to deal with transient congestion. See: * https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1559.md * https://eips.ethereum.org/EIPS/eip-1559 """ stage = Stage(previous_state["stage"]) if stage not in [Stage.EIP1559, Stage.PROOF_OF_STAKE]: return { "base_fee_per_gas": 0, "total_base_fee": 0, "total_priority_fee_to_miners": 0, "total_priority_fee_to_validators": 0, } # Parameters dt = params["dt"] gas_target_process = params["gas_target_process"] # Gas ELASTICITY_MULTIPLIER = params["ELASTICITY_MULTIPLIER"] base_fee_process = params["base_fee_process"] priority_fee_process = params["priority_fee_process"] # State Variables run = previous_state["run"] timestep = previous_state["timestep"] # Get samples for current run and timestep from base fee, priority fee, and transaction processes base_fee_per_gas = base_fee_process(run, timestep * dt) # Gwei per Gas gas_target = gas_target_process(run, timestep * dt) # Gas # Ensure basefee changes by no more than 1 / BASE_FEE_MAX_CHANGE_DENOMINATOR % _BASE_FEE_MAX_CHANGE_DENOMINATOR = params["BASE_FEE_MAX_CHANGE_DENOMINATOR"] # assert ( # abs(basefee - previous_basefee) / previous_basefee # <= constants.slots_per_epoch / BASE_FEE_MAX_CHANGE_DENOMINATOR # if timestep > 1 # else True # ), "basefee changed by more than 1 / BASE_FEE_MAX_CHANGE_DENOMINATOR %" avg_priority_fee_per_gas = priority_fee_process(run, timestep * dt) # Gwei per Gas if stage in [Stage.EIP1559]: gas_used = constants.pow_blocks_per_epoch * gas_target # Gas else: # stage is Stage.PROOF_OF_STAKE gas_used = constants.slots_per_epoch * gas_target # Gas # Calculate total base fee, and priority fee to validators total_base_fee = gas_used * base_fee_per_gas # Gwei total_priority_fee = gas_used * avg_priority_fee_per_gas # Gwei if stage in [Stage.PROOF_OF_STAKE]: total_priority_fee_to_miners = 0 total_priority_fee_to_validators = total_priority_fee else: total_priority_fee_to_miners = total_priority_fee total_priority_fee_to_validators = 0 # Check if the block used too much gas assert ( gas_used <= gas_target * ELASTICITY_MULTIPLIER * constants.slots_per_epoch ), "invalid block: too much gas used" return { "base_fee_per_gas": base_fee_per_gas, "total_base_fee": total_base_fee * dt, "total_priority_fee_to_miners": total_priority_fee_to_miners * dt, "total_priority_fee_to_validators": total_priority_fee_to_validators * dt, }
def policy_network_issuance(params, substep, state_history, previous_state) ‑> Dict[str, float]
-
Network Issuance Policy Function
Calculate the total network issuance and issuance from Proof of Work block rewards.
Expand source code
def policy_network_issuance( params, substep, state_history, previous_state ) -> typing.Dict[str, ETH]: """ ## Network Issuance Policy Function Calculate the total network issuance and issuance from Proof of Work block rewards. """ # Parameters dt = params["dt"] daily_pow_issuance = params["daily_pow_issuance"] # State Variables stage = previous_state["stage"] amount_slashed = previous_state["amount_slashed"] total_base_fee = previous_state["total_base_fee"] total_priority_fee_to_validators = previous_state[ "total_priority_fee_to_validators" ] total_online_validator_rewards = previous_state["total_online_validator_rewards"] # Calculate network issuance in ETH network_issuance = ( # Remove priority fee to validators which is not issuance (ETH transferred rather than minted) (total_online_validator_rewards - total_priority_fee_to_validators) - amount_slashed - total_base_fee ) / constants.gwei # Calculate Proof of Work issuance pow_issuance = ( daily_pow_issuance / constants.epochs_per_day if Stage(stage) in [Stage.BEACON_CHAIN, Stage.EIP1559] else 0 ) network_issuance += pow_issuance * dt return { "network_issuance": network_issuance, "pow_issuance": pow_issuance, }
def policy_upgrade_stages(params, substep, state_history, previous_state)
-
Upgrade Stages Policy
Transitions the model from one stage in the Ethereum network upgrade process to the next at different milestones.
This is essentially a finite-state machine: https://en.wikipedia.org/wiki/Finite-state_machine
Expand source code
def policy_upgrade_stages(params, substep, state_history, previous_state): """ ## Upgrade Stages Policy Transitions the model from one stage in the Ethereum network upgrade process to the next at different milestones. This is essentially a finite-state machine: https://en.wikipedia.org/wiki/Finite-state_machine """ # Parameters dt = params["dt"] stage: Stage = params["stage"] date_start = params["date_start"] date_eip1559 = params["date_eip1559"] date_pos = params["date_pos"] # State Variables current_stage = previous_state["stage"] timestep = previous_state["timestep"] # Calculate current timestamp from timestep timestamp = date_start + datetime.timedelta( days=(timestep * dt / constants.epochs_per_day) ) # Initialize stage State Variable at start of simulation if current_stage is None: current_stage = stage else: # Convert Stage enum value (int) to Stage enum current_stage = Stage(current_stage) # Stage finite-state machine if stage == Stage.ALL: # If Stage ALL selected, transition through all stages # at different timestamps if ( current_stage in [Stage.ALL, Stage.BEACON_CHAIN] and timestamp < date_eip1559 ): current_stage = Stage.BEACON_CHAIN elif ( current_stage in [Stage.BEACON_CHAIN, Stage.EIP1559] and timestamp < date_pos ): current_stage = Stage.EIP1559 else: current_stage = Stage.PROOF_OF_STAKE elif stage == Stage.BEACON_CHAIN: # If Stage BEACON_CHAIN selected, only execute single stage current_stage = Stage.BEACON_CHAIN elif stage == Stage.EIP1559: # If Stage EIP1559 selected, only execute single stage current_stage = Stage.EIP1559 elif stage == Stage.PROOF_OF_STAKE: # If Stage PROOF_OF_STAKE selected, only execute single stage current_stage = Stage.PROOF_OF_STAKE else: # Else, raise exception if invalid Stage raise Exception("Invalid Stage selected") return { "stage": current_stage.value, "timestamp": timestamp, }
def update_eth_price(params, substep, state_history, previous_state, policy_input) ‑> Tuple[str, float]
-
ETH Price State Update Function
Update the ETH price from the
eth_price_process
.Expand source code
def update_eth_price( params, substep, state_history, previous_state, policy_input ) -> typing.Tuple[str, USD_per_ETH]: """ ## ETH Price State Update Function Update the ETH price from the `eth_price_process`. """ # Parameters dt = params["dt"] eth_price_process = params["eth_price_process"] # State Variables run = previous_state["run"] timestep = previous_state["timestep"] # Get the ETH price sample for the current run and timestep eth_price_sample = eth_price_process(run, timestep * dt) return "eth_price", eth_price_sample
def update_eth_supply(params, substep, state_history, previous_state, policy_input) ‑> Tuple[str, float]
-
ETH Supply State Update Function
Update the ETH supply from the Network Issuance Policy Function.
Expand source code
def update_eth_supply( params, substep, state_history, previous_state, policy_input ) -> typing.Tuple[str, ETH]: """ ## ETH Supply State Update Function Update the ETH supply from the Network Issuance Policy Function. """ # Policy Inputs network_issuance = policy_input["network_issuance"] # State variables eth_supply = previous_state["eth_supply"] return "eth_supply", eth_supply + network_issuance