Calculate Activation Epoch

get_queue_position.py

#!/usr/bin/env python3
import sys
import requests_cache
import json

BEACON_ENDPOINT = "http://localhost:5052"

FAR_FUTURE_EPOCH = 18446744073709551615
MAX_EFFECTIVE_BALANCE = 32000000000
MIN_PER_EPOCH_CHURN_LIMIT = 4
CHURN_LIMIT_QUOTIENT = 65536
SECONDS_PER_SLOT = 12
SLOTS_PER_EPOCH = 32
MAX_SEED_LOOKAHEAD = 4

# modified once
active_validators = 0
finalized_epoch = 0
all_validators = []

def get_validators_url():
    return '{}/eth/v1/beacon/states/head/validators'.format(BEACON_ENDPOINT)

def get_finality_url():
    return '{}/eth/v1/beacon/states/head/finality_checkpoints'.format(BEACON_ENDPOINT)

def get_response(url):
    session = requests_cache.CachedSession('beacon_response', expire_after=1800)
    return session.get(url, headers={'Accept': 'application/json'}).json()

def get_all_validators():
    global all_validators
    if all_validators:
        return all_validators
    response = get_response(get_validators_url())
    all_validators = response["data"]
    return all_validators

def get_finalized_epoch():
    global finalized_epoch
    if finalized_epoch:
        return finalized_epoch
    response = get_response(get_finality_url())
    finalized_epoch = int(response["data"]["finalized"]["epoch"])
    return finalized_epoch

def get_active_validators():
    global active_validators
    if active_validators:
        return active_validators
    validators = get_all_validators()
    n_active = 0
    for v in validators:
        if is_active_validator(v):
            n_active += 1
    active_validators = n_active
    return active_validators

def is_active_validator(validator):
    epoch = get_finalized_epoch() + 2
    return (int(validator["validator"]["activation_epoch"]) <= epoch < int(validator["validator"]["exit_epoch"]))

def compute_activation_exit_epoch(epoch):
    return epoch + 1 + MAX_SEED_LOOKAHEAD

def is_eligible_for_activation(validator, finalized_epoch):
    return (
        int(validator["validator"]["activation_eligibility_epoch"]) <= finalized_epoch
        and int(validator["validator"]["activation_epoch"]) == FAR_FUTURE_EPOCH
    )

def get_validator_churn_limit():
    n_active_validator_indices = get_active_validators()
    return max(MIN_PER_EPOCH_CHURN_LIMIT, n_active_validator_indices // CHURN_LIMIT_QUOTIENT)

def get_pending_validators():
    finalized_epoch = get_finalized_epoch()
    validators = get_all_validators()
    result = []
    print("Total Validators: {}".format(len(validators)))
    print("Active Validators: {}".format(get_active_validators()))
    for validator in validators:
        if is_eligible_for_activation(validator, finalized_epoch):
            result.append(validator)

    result.sort(key=lambda v: (int(v["validator"]["activation_eligibility_epoch"]), int(v["index"])))    
    return result

def get_remaining_epochs(position):
    n_epochs = MAX_SEED_LOOKAHEAD
    n_active_validator_indices = get_active_validators()
    while position > 0:
        churn_limit = max(MIN_PER_EPOCH_CHURN_LIMIT, n_active_validator_indices // CHURN_LIMIT_QUOTIENT)
        n_active_validator_indices += min(churn_limit, position)
        position = max(position - churn_limit, 0)
        n_epochs += 1
    return n_epochs

def print_time_remaining(seconds):
    days, remainder  = divmod(seconds, 86400)
    hours, remainder = divmod(remainder, 3600)
    minutes, seconds = divmod(remainder, 60)
    print("Time Remaining:", end = " ")
    if days:
        print("{} days".format(days), end = " ")
    if hours:
        print("{} hours".format(hours), end = " ")
    if minutes:
        print("{} minutes".format(minutes), end = "")
    print("")

def print_validator_queue_position(index):
    pending_validators = get_pending_validators()
    current_epoch = get_finalized_epoch() + 2
    print("Total Pending Validators: {}".format(len(pending_validators) + MAX_SEED_LOOKAHEAD * get_validator_churn_limit()))
    for pos, v in enumerate(pending_validators):
        if v["index"] == index:
            print("Validator found in Activation Queue")
            print(json.dumps(v, indent=2))
            print("Position: {}".format(pos))
            remaining_epochs = get_remaining_epochs(pos)
            print("Activation Epoch:", current_epoch + remaining_epochs)
            print_time_remaining(remaining_epochs * SECONDS_PER_SLOT * SLOTS_PER_EPOCH)
            sys.exit(0)
    print("Validator {} not found in Activation Queue".format(index))

def main():
    if len(sys.argv) < 2:
        print("Usage: {} [VALIDATOR_INDEX]".format(sys.argv[0]))
        sys.exit(1)
    print_validator_queue_position(sys.argv[1])
main()