medalla-1-data-prep.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "---\n",
    "# front matter for Jekyll\n",
    "title: \"Data Pre-processing for the Medalla Data Challenge\"\n",
    "permalink: \"/posts/medalla-data-prep/\"\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This article documents the pre-processing steps carried out for an entry into the [Medalla data challenge](https://ethereum.org/en/eth2/get-involved/medalla-data-challenge/). Like all the articles in this series, it is a conversion of a Jupyter notebook — the originals of which are saved in [this GitHub respository](https://github.com/pintail-xyz/medalla_analysis).\n",
    "\n",
    " Visualisations and analysis of this data can be found on the following pages:\n",
    "\n",
    "- [Medalla Participation Rates: A Validator Taxonomy](/posts/medalla-validator-taxonomy)\n",
    "- [The Medalla Network Under Stress](/posts/medalla-network-stress)\n",
    "- [Eth2 Client Comparisons from Medalla Data](/posts/medalla-client-comparison)\n",
    "\n",
    "The data used is taken from a [database dump](http://mdc.mcdee.net/chain-487600.dmp), kindly shared by Jim McDonald ([@jgm](https://twitter.com/jgm)) of [Attestant](https://www.attestant.io/), which includes beacon chain data for the first approximately 15,000 epochs (up to around 11 October 2020). The schemas for the tables in this database are included in the separate [schemas](schemas) file. Jim has made his `chaind` process for extracting beacon chain data from an eth2 client [available for use](https://github.com/wealdtech/chaind), so the results in this article can be replicated and updated by first generating a database from `chaind` and your eth2 node of choice.\n",
    "\n",
    "Special thanks also to Ben Edgington ([@benjaminion_xyz](https://twitter.com/benjaminion_xyz)) whose [Eth2 Annotated Spec](https://benjaminion.xyz/eth2-annotated-spec/phase0/beacon-chain/) has been a vital resource in understanding how the beacon chain works.\n",
    "\n",
    "## Setup\n",
    "First up, we import the Python libraries to be used in this analysis, connect to the `chaind` database and pull out some basic information about the data we have saved there:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [],
   "source": [
    "# imports\n",
    "import time\n",
    "\n",
    "import psycopg2\n",
    "from bitlist import bitlist\n",
    "from IPython.display import clear_output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [],
   "source": [
    "# open/restart connection to chaind database\n",
    "try:\n",
    "    cursor.close()\n",
    "    connection.close()\n",
    "except:\n",
    "    pass\n",
    "\n",
    "connection = psycopg2.connect(user=\"chain\", host=\"127.0.0.1\", database=\"chain\", password=\"medalla\")\n",
    "cursor = connection.cursor()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "latest slot: 487600, latest complete epoch: 15236\n",
      "latest block root: 5a36cf6aee95f69a79e44ba6bcb8f81846d1704dc05e74e8e467c37fca4c29ca\n"
     ]
    }
   ],
   "source": [
    "# basic info about the dataset\n",
    "cursor.execute(\"SELECT MAX(f_slot) FROM t_blocks\")\n",
    "latest_slot = cursor.fetchone()[0]\n",
    "n_slots = latest_slot + 1\n",
    "n_epochs = (n_slots - (n_slots % 32)) // 32\n",
    "print(f\"latest slot: {latest_slot}, latest complete epoch: {n_epochs - 1}\")\n",
    "\n",
    "cursor.execute(\"SELECT f_slot, f_root FROM t_blocks ORDER BY f_slot DESC LIMIT 1\")\n",
    "latest_block = cursor.fetchone()\n",
    "slot, root = latest_block[0], latest_block[1].hex()\n",
    "print(f\"latest block root: {root}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Database Modifications\n",
    "Next we make a few changes to the postgresql database provided by `chaind`. The database includes all blocks observed by Jim's Teku node, including a number of *orphaned* blocks which did not become *canonical*, that is, part of the finalised chain. Working out which blocks are canonical and which are orphaned will be valuable later, so we create extra columns in the `t_blocks` and `t_attestations` tables to mark those blocks and the attestations which were included in them. We also create a new index on `t_attestations` to allow us to select attestations according to the hash of the block in which they were included.\n",
    "\n",
    "We then add a new table, `t_validator_performance`, which we will populate later with information about how each validator performed for each of its assigned attestation duties.\n",
    "\n",
    "Finally we modify the `t_validators` table, adding some more columns we will populate with information about when they started and finished attesting, and how many blocks they succesfully proposed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "completed database modifications in 00:02:39\n"
     ]
    }
   ],
   "source": [
    "# database modifications required for this analysis\n",
    "\n",
    "start_time = time.time()\n",
    "\n",
    "print(\"adding f_canonical column to t_blocks table\", end='\\r')\n",
    "cursor.execute(\"ALTER TABLE t_blocks DROP COLUMN IF EXISTS f_canonical\")\n",
    "cursor.execute(\"ALTER TABLE t_blocks ADD COLUMN f_canonical BOOLEAN DEFAULT false\")\n",
    "\n",
    "print(\"adding f_canonical and f_correct columns to t_attestations table\", end='\\r')\n",
    "cursor.execute(\"ALTER TABLE t_attestations DROP COLUMN IF EXISTS f_canonical\")\n",
    "cursor.execute(\"ALTER TABLE t_attestations ADD COLUMN f_canonical BOOLEAN DEFAULT false\")\n",
    "\n",
    "print(\"adding i_inclusion_block_root index to t_attestations table     \", end='\\r')\n",
    "cursor.execute(\"DROP INDEX IF EXISTS i_inclusion_block_root\")\n",
    "cursor.execute(\"CREATE INDEX i_inclusion_block_root ON t_attestations (f_inclusion_block_root)\")\n",
    "\n",
    "print(\"creating t_validator_performance table                          \", end='\\r')\n",
    "cursor.execute(\"DROP TABLE IF EXISTS t_validator_performance\")\n",
    "cursor.execute(\"CREATE TABLE t_validator_performance (f_slot bigint NOT NULL PRIMARY KEY, \"\n",
    "               \"f_validator_index int[], f_performance int[], f_effectiveness float[], f_correct int[])\")\n",
    "\n",
    "print(\"adding additional columns to t_validators table                 \", end='\\r')\n",
    "cursor.execute(\"ALTER TABLE t_validators \"\n",
    "               \"DROP COLUMN IF EXISTS f_first_attested_epoch, \"\n",
    "               \"DROP COLUMN IF EXISTS f_latest_attested_epoch, \"\n",
    "               \"DROP COLUMN IF EXISTS f_client, \"\n",
    "               \"DROP COLUMN IF EXISTS f_proposed_count\")\n",
    "cursor.execute(\"ALTER TABLE t_validators \"\n",
    "               \"ADD COLUMN f_first_attested_epoch int DEFAULT -1, \"\n",
    "               \"ADD COLUMN f_latest_attested_epoch int DEFAULT -1, \"\n",
    "               \"ADD COLUMN f_client text, \"\n",
    "               \"ADD COLUMN f_proposed_count int\")\n",
    "\n",
    "connection.commit()\n",
    "\n",
    "elapsed = time.strftime(\"%H:%M:%S\",time.gmtime(time.time() - start_time))\n",
    "clear_output()\n",
    "print(f\"completed database modifications in {elapsed}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Identifying Canonical Blocks\n",
    "One of the properties of eth2's [Casper FFG](https://arxiv.org/pdf/1710.09437.pdf) finality process is that blocks can be *finalised*, that is, they become unrevertable. Over short timescales it is possible for the eth2 beacon chain to fork, just as with existing Proof-of-Work blockchains (including the existing eth1 chain). However, with sufficient validator participation (representing at least ⅔ of the total amount of staked ether), eventually the validator set reaches consensus on one *canonical* chain.\n",
    "\n",
    "Also as in other blockchains, each block contains a hash of a *parent* — that is, a block at an earlier time identified as the previous block in the chain. Identiyfing all blocks in the canonical chain is therefore simply a matter of taking a recently-finalised block and tracking back through the chain via each block's parents. Helpfully, the most recent block in the `chaind` database used in this analysis *was* finalised, according to [beaconcha.in](https://beaconcha.in), so we will start tracking back from there. Any block not lying on this chain will be considered to be an orphan."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "finished (reached back to slot 0) in 00:33:26.\n",
      "database contains 358,048 blocks of which 349,856 are canonical (2.3% orphan blocks)\n"
     ]
    }
   ],
   "source": [
    "# identify canonical blocks\n",
    "\n",
    "# block 5a36cf6aee95f69a79e44ba6bcb8f81846d1704dc05e74e8e467c37fca4c29ca indicated as finalised on beaconcha.in.\n",
    "# consider all ancestors of this block to be canonical.\n",
    "\n",
    "start_time = time.time()\n",
    "\n",
    "while True:\n",
    "    # mark current block as canonical\n",
    "    cursor.execute(f\"UPDATE t_blocks SET f_canonical = true WHERE f_root = '\\\\x{root}'\")\n",
    "    \n",
    "    # mark attestations from this block as canonical\n",
    "    cursor.execute(f\"UPDATE t_attestations SET f_canonical = true WHERE f_inclusion_block_root = '\\\\x{root}'\")\n",
    "    \n",
    "    # identify parent block\n",
    "    cursor.execute(f\"SELECT f_slot, f_parent_root FROM t_blocks WHERE f_root = '\\\\x{root}'\")\n",
    "    result = cursor.fetchone()\n",
    "    \n",
    "    # exit loop if this block's parent is not in database (or slot=0 in which case no parent)\n",
    "    if not result:\n",
    "        break\n",
    "    \n",
    "    slot, root = result[0], result[1].hex()\n",
    "    \n",
    "    # show progress info\n",
    "    if slot % 32 == 0:\n",
    "        seconds = time.time() - start_time\n",
    "        elapsed = time.strftime(\"%H:%M:%S\",time.gmtime(seconds))\n",
    "        done = 1 - (slot+1) / n_slots\n",
    "        left = time.strftime(\"%H:%M:%S\",time.gmtime(seconds * (1 / done - 1)))\n",
    "        print(f\"working backwards through blocks / current slot: {slot} ({100*done:.2f}% complete) / \"\n",
    "              f\"{elapsed} elapsed / {left} left     \", end='\\r')\n",
    "\n",
    "connection.commit()\n",
    "\n",
    "clear_output()\n",
    "print(f\"finished (reached back to slot {slot:,}) in {elapsed}.\")\n",
    "\n",
    "cursor.execute(\"SELECT COUNT(*), SUM(f_canonical::int) FROM t_blocks\")\n",
    "result = cursor.fetchone()\n",
    "orphan_ratio = 1 - result[1] / result[0]\n",
    "print(f\"database contains {result[0]:,} blocks of which {result[1]:,} are canonical \"\n",
    "      F\"({100*orphan_ratio:.1f}% orphan blocks)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [],
   "source": [
    "# (this code is an optimisation for the calculations below but has not been used)\n",
    "\n",
    "# identify chain head for each slot\n",
    "head = []\n",
    "for slot in range(n_slots):\n",
    "    cursor.execute(f\"SELECT f_root FROM t_blocks WHERE f_slot = {slot} AND f_canonical\")\n",
    "    result = cursor.fetchone()\n",
    "    if result:\n",
    "        head_slot, head_root = slot, result[0].hex()\n",
    "    head.append([head_slot, head_root])\n",
    "\n",
    "# find the gap to the next canonical block at each slot\n",
    "gap = 1\n",
    "for i, h in enumerate(head):\n",
    "    while i+gap < len(head) and head[i+gap][0] != i+gap:\n",
    "        gap += 1\n",
    "    h.append(gap)\n",
    "    gap = 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "database contains 22,723,853 attestations of which 22,246,751 are canonical (2.1% from orphan blocks)\n"
     ]
    }
   ],
   "source": [
    "# check proportion of attestations from canoncial blocks broadly matches block orphan ratio\n",
    "cursor.execute(\"SELECT COUNT(*), SUM(f_canonical::int) FROM t_attestations\")\n",
    "result = cursor.fetchone()\n",
    "\n",
    "print(f\"database contains {result[0]:,} attestations of which {result[1]:,} are canonical \"\n",
    "      f\"({100 * (1 - result[1] / result[0]):.1f}% from orphan blocks)\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Blockchain Time\n",
    "For the eth2 beacon chain, the base unit of time is a *slot*. Slots occur regularly every 12 seconds, and in each slot a single canonical block may be produced (but some slots may be *empty* if no block for that slot is included in the finalised chain). A set of 32 slots is called an *epoch*, and in each epoch every active validator is expected to produce a single attestation. \n",
    "\n",
    "## Decoding Attestations\n",
    "The majority of the analysis of the Medalla data will be concerned with *attestations*, that is, cryptographically-signed votes by individual validators on blocks they have evaluated and found to be valid. These attestations themselves constitute the bulk of the data included in subsequent blocks. The beacon chain's *aggregation* of these attestations and their signatures is one of the key advances which allows eth2 to include many validators in the consensus process, and therefore promote decentralisation. For the purposes of analysis however, we need to unpack aggregated attestations to understand the performance of individual validators.\n",
    "\n",
    "Each active validator is allocated to a *committee* whose attestations will be comnined into a single aggregated attestation. Information on which validators were part of each committee is included in the `chaind` database `t_beacon_committees` table. The attestations themselves are saved in the `t_attestations` table, which includes a column, `f_aggregation_bits`, which is effectively a series of single-bit-flags indicating which validators were aggregated into the final attestation.\n",
    "\n",
    "Ultimately different versions of aggregated attestations end up being included in the beacon chain, at different slots. Accordingly, an individual validator may appear in several of these attestations. How many slots after the the block being produced a validator manages to get their *first* attestation included may be thought of as measure of their performance — the sooner the attestation is included, the better (validator rewards also recognise this). This measure of performance on an individual attestation is called *inclusion distance* where an inclusion distance of 0 indicates the attestation was included at the very next slot, and an inclusion distance of 1 is the slot after (etc.).\n",
    "\n",
    "The below cell calculates the *first inclusion distance* for each active validator, at each epoch. If the validator does not have an attestation included at all, the attestation is said to be *missed*, denoted by an inclusion distance of -1. We also calculate the [attestation effectiveness](https://www.attestant.io/posts/defining-attestation-effectiveness/) for each validator, as the inverse of the *adjusted inclusion distance*. Finally, we record whether each validator attested *correctly* — that is, did the block it voted for end up being finalised?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "attestations processed in 02:23:47\n",
      "indexing validator performance table...done\n",
      "saving validator first/latest attestation epochs...done\n"
     ]
    }
   ],
   "source": [
    "# decode attestations, calculate performance, identify first/latest attestation for each validator\n",
    "\n",
    "def decode_attestation(aggregation_bits, committee):\n",
    "    n_validators = len(committee)\n",
    "    aggregation_bitlist = bitlist(aggregation_bits.tobytes()[::-1])[:-(n_validators+1):-1]\n",
    "    attested = [True if bit == 1 else False for bit in aggregation_bitlist]\n",
    "    return attested\n",
    "\n",
    "cursor.execute(\"SELECT COUNT(*) FROM t_validators\")\n",
    "n_validators = cursor.fetchone()[0]\n",
    "validator_info = [{\"first_epoch\":  -1, \"latest_epoch\": -1} for i in range(n_validators)]\n",
    "\n",
    "start_time = time.time()\n",
    "#last_update = 0\n",
    "\n",
    "for slot in range(latest_slot):\n",
    "    epoch = slot // 32\n",
    "    # get committees of validators assigned for attestation this slot\n",
    "    cursor.execute(f\"SELECT f_committee FROM t_beacon_committees WHERE f_slot = {slot} ORDER BY f_index\")\n",
    "    result = cursor.fetchall()\n",
    "    # committee_lookup is a list of committees (themselves lists of validator indices) for this slot\n",
    "    committee_lookup = [result[committee_index][0] for committee_index in range(len(result))]\n",
    "    # committee_performance is the inclusion distance if committee member attested, -1 otherwise\n",
    "    committee_performance   = [[-1] * len(committee) for committee in committee_lookup]\n",
    "    # committee_effectiveness is inverse of the adjusted inclusion distance (accounting for empty slots)\n",
    "    committee_effectiveness = [[0]  * len(committee) for committee in committee_lookup]\n",
    "    # committee_correct says whether each member of the committee voted for a canonical block\n",
    "    committee_correct       = [[0]  * len(committee) for committee in committee_lookup]\n",
    "    \n",
    "    # work out when the first attestation opportunity for this slot was\n",
    "    cursor.execute(f\"SELECT MIN(f_slot) FROM t_blocks WHERE f_slot > {slot} AND f_canonical = true\")\n",
    "    earliest_inclusion_slot = cursor.fetchone()[0]\n",
    "    min_distance = earliest_inclusion_slot - slot - 1\n",
    "    \n",
    "    # get the chain head at this slot\n",
    "    cursor.execute(f\"SELECT f_slot, f_root FROM t_blocks WHERE f_slot <= {slot} AND f_canonical \"\n",
    "                   f\"ORDER BY f_slot DESC LIMIT 1\")\n",
    "    result = cursor.fetchone()\n",
    "    head_slot, head_root = result[0], result[1].hex()\n",
    "    \n",
    "    # get all canonical attestations made for this slot\n",
    "    cursor.execute(f\"SELECT f_committee_index, f_inclusion_slot, f_aggregation_bits, \"\n",
    "                   f\"f_inclusion_block_root, f_beacon_block_root FROM t_attestations \"\n",
    "                   f\"WHERE f_slot = {slot} AND f_canonical ORDER BY f_inclusion_slot\")\n",
    "    attestations = cursor.fetchall()\n",
    "    \n",
    "    for attestation in attestations:\n",
    "        committee_index, inclusion_slot, aggregation_bits = attestation[:3]\n",
    "        inclusion_root, attestation_head_root = attestation[3].hex(), attestation[4].hex()\n",
    "        \n",
    "        correct = 1 if attestation_head_root == head_root else -1\n",
    "        \n",
    "        inclusion_distance = inclusion_slot - slot - 1\n",
    "        committee_participation = decode_attestation(aggregation_bits, committee_lookup[committee_index])\n",
    "        \n",
    "        # record the shortest inclusion_distance for each member\n",
    "        for position, participated in enumerate(committee_participation):\n",
    "            if participated and (committee_performance[committee_index][position] == -1):\n",
    "                committee_performance[committee_index][position] = inclusion_distance\n",
    "                committee_effectiveness[committee_index][position] = 1 / (1 + inclusion_distance - min_distance)\n",
    "                committee_correct[committee_index][position] = correct\n",
    "        \n",
    "    # flatten lookup tables\n",
    "    validators_flat    = [el for committee in committee_lookup        for el in committee]\n",
    "    performance_flat   = [el for committee in committee_performance   for el in committee]\n",
    "    effectiveness_flat = [el for committee in committee_effectiveness for el in committee]\n",
    "    correct_flat       = [el for committee in committee_correct       for el in committee]\n",
    "    cursor.execute(\"INSERT INTO t_validator_performance VALUES (%s, %s, %s, %s, %s)\",\n",
    "                       (slot, validators_flat, performance_flat, effectiveness_flat, correct_flat))\n",
    "\n",
    "    # save attestation performance/effectiveness and update first/latest attestation info\n",
    "    for i, validator in enumerate(validators_flat):\n",
    "        if performance_flat[i] != -1:\n",
    "            validator_info[validator]['latest_epoch'] = epoch\n",
    "            if validator_info[validator]['first_epoch'] == -1:\n",
    "                validator_info[validator]['first_epoch'] = epoch\n",
    "            \n",
    "    seconds = time.time() - start_time\n",
    "    elapsed = time.strftime(\"%H:%M:%S\",time.gmtime(seconds))\n",
    "    left = time.strftime(\"%H:%M:%S\",time.gmtime(seconds * (n_slots / (slot+1)-1)))\n",
    "    percentage = 100*(slot+1)/n_slots\n",
    "    \n",
    "    print(f\"processing attestations: epoch {epoch} of {n_epochs} ({percentage:.2f}%) / \"\n",
    "          f\"{elapsed} elapsed / {left} left       \", end='\\r')\n",
    "\n",
    "clear_output()\n",
    "print(f\"attestations processed in {elapsed}\")\n",
    "\n",
    "print(\"indexing validator performance table...\", end='')\n",
    "cursor.execute(\"CREATE INDEX i_epoch ON t_validator_performance (f_slot)\")\n",
    "print(\"done\")\n",
    "\n",
    "print(\"saving validator first/latest attestation epochs...\", end='')\n",
    "for validator_index, info in enumerate(validator_info):\n",
    "    cursor.execute(f\"UPDATE t_validators \"\n",
    "                   f\"SET f_first_attested_epoch = {info['first_epoch']}, \"\n",
    "                   f\"   f_latest_attested_epoch = {info['latest_epoch']}\"\n",
    "                   f\"WHERE f_index = {validator_index}\")\n",
    "print(\"done\")\n",
    "\n",
    "connection.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Identifying Validator Clients and Block Producers\n",
    "There is no definitive way to identify which client software powers each validator — as long as a given client follows the protocol rules, it is equivalent to any other client from the perspective of the network. However, for the purposes of Medalla, many users chose to indicate what client they were using in the `graffiti` field in blocks they propose. Those aiming to collect [POAP non-fungible tokens (NFTs)](https://beaconcha.in/poap) will have set their validators to include a graffiti string indicating their eth1 deposit address, and ending with a letter A, B, C, D or E, corresponding to the Prysm, Lighthouse, Teku, Nimbus and Lodestar clients respectively. For example, the third block in the dataset has the graffiti `poapg4eM7/cwRi/ZhaZg0zp6b9A6JlcA` indicating the block was produced by the Prysm client. Other validators have included the full name of their client in the graffiti — the very next block includes the graffiti `Lighthouse/v0.2.0/f26adc0a`. It seems a reasonable guess that this block was produced by the Lighthouse client.\n",
    "\n",
    "Of course, the graffiti field can be set to whatever the user running the validator chooses, including misleading information about the client being used. For the most part there is little reason to expect users would be untruthful, with the exception that some may have chosen to game the POAP system to collect NFTs for all clients without the hassle of actually installing and using them. Nonetheless, to make progress with our analysis we will need to assume that the majority of validators were honest. Where we have large numbers of validators indicating the use of a certain client therefore, we might expect to gain  some insight into how that client has performed in aggregate. We may need to be a little more cautious if there are clients with a very small number of validators indicating participation through their beacon chain graffiti.\n",
    "\n",
    "For the purposes of this analysis, we will assume that any validator which consistently indicates a particular client through its graffiti is genuinely running the claimed client. Validators which appear to switch clients, or which refer to mulitple clients in their graffiti, will be excluded (NB This will exclude even users who genuinely have switched clients, since validators produce blocks only occasionally, and it would be impossible to know exactly when they switched).\n",
    "\n",
    "The cell below iterates through all canonical blocks in the dataset and where available, uses the `graffiti` field to guess which client produced the blocks. If information from different blocks suggests different clients were used, the validator is marked as *ambiguous*. We also count up the total number of blocks proposed by each validator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "code_folding": [
     0
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "block proposers processed in 00:00:01\n",
      "saving validator info...done\n"
     ]
    }
   ],
   "source": [
    "# determine validator clients from block graffiti, count proposed blocks per validator\n",
    "\n",
    "clients = [''] * n_validators\n",
    "block_counts = [0] * n_validators\n",
    "\n",
    "cursor.execute(f\"SELECT f_proposer_index, f_graffiti FROM t_blocks WHERE f_canonical = true\")\n",
    "proposer_info = cursor.fetchall()\n",
    "\n",
    "start_time = time.time()\n",
    "last_update = 0\n",
    "\n",
    "for i, info in enumerate(proposer_info):\n",
    "    validator_index, graffiti = info[0], info[1].tobytes().decode(errors='ignore').lower()\n",
    "    \n",
    "    block_counts[validator_index] += 1\n",
    "    \n",
    "    prior = clients[validator_index]\n",
    "    \n",
    "    if prior == \"ambiguous\":\n",
    "        continue\n",
    "\n",
    "    pr_flag = (graffiti[:4] == \"poap\" and graffiti[-1] == \"a\") or graffiti.find(\"prysm\")      != -1\n",
    "    li_flag = (graffiti[:4] == \"poap\" and graffiti[-1] == \"b\") or graffiti.find(\"lighthouse\") != -1\n",
    "    te_flag = (graffiti[:4] == \"poap\" and graffiti[-1] == \"c\") or graffiti.find(\"teku\")       != -1\n",
    "    ni_flag = (graffiti[:4] == \"poap\" and graffiti[-1] == \"d\") or graffiti.find(\"nimbus\")     != -1\n",
    "    lo_flag = (graffiti[:4] == \"poap\" and graffiti[-1] == \"e\") or graffiti.find(\"lodestar\")   != -1\n",
    "    n_flags = pr_flag + li_flag + te_flag + ni_flag + lo_flag\n",
    "    \n",
    "    if n_flags > 1:\n",
    "        clients[validator_index] = \"ambiguous\"\n",
    "    elif pr_flag:\n",
    "        clients[validator_index] = \"prysm\"      if prior in [\"\", \"prysm\"]      else \"ambiguous\"\n",
    "    elif li_flag:\n",
    "        clients[validator_index] = \"lighthouse\" if prior in [\"\", \"lighthouse\"] else \"ambiguous\"\n",
    "    elif te_flag:\n",
    "        clients[validator_index] = \"teku\"       if prior in [\"\", \"teku\"]       else \"ambiguous\"\n",
    "    elif ni_flag:\n",
    "        clients[validator_index] = \"nimbus\"     if prior in [\"\", \"nimbus\"]     else \"ambiguous\"\n",
    "    elif lo_flag:\n",
    "        clients[validator_index] = \"lodestar\"   if prior in [\"\", \"lodestar\"]   else \"ambiguous\"\n",
    "        \n",
    "    # show progress info\n",
    "    if time.time() - last_update > 0.05:\n",
    "        last_update = time.time()\n",
    "        seconds = time.time() - start_time\n",
    "        elapsed = time.strftime(\"%H:%M:%S\",time.gmtime(seconds))\n",
    "        done = (i+1) / len(proposer_info)\n",
    "        left = time.strftime(\"%H:%M:%S\",time.gmtime(seconds * (1 / done - 1)))\n",
    "        print(f\"working through blocks / {100*done:.2f}% complete / \"\n",
    "              f\"{elapsed} elapsed / {left} left\", end='\\r')\n",
    "\n",
    "clear_output()\n",
    "print(f\"block proposers processed in {elapsed}\")\n",
    "print(\"saving validator info...\", end='')\n",
    "for i, client in enumerate(clients):\n",
    "    cursor.execute(\"UPDATE t_validators SET f_client = %s, f_proposed_count = %s\"\n",
    "                   \"WHERE f_index = %s\", (client, block_counts[i], i))\n",
    "\n",
    "connection.commit()\n",
    "print(\"done\")"
   ]
  }
 ],
 "metadata": {
  "jekyll": {
   "layout": "single",
   "permalink": "/medalla-data-prep/",
   "title": "Data Pre-processing for the Medalla Data Challenge"
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}