Module model.parts.ethereum_system

Ethereum System

General Ethereum mechanisms, such as managing the system upgrade process, the EIP1559 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 EIP1559 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