get_emit.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import bittensor\n",
    "import pandas as pd\n",
    "import json\n",
    "import substrateinterface as pysub\n",
    "\n",
    "from typing import List, Tuple, Optional, Dict\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "sub = bittensor.subtensor(\"wss://archive.chain.opentensor.ai:443\")\n",
    "BLOCK_TIME = 12\n",
    "\n",
    "BLOCKS_PER_HOUR = 3600 / BLOCK_TIME"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "START_BLOCK = 3_791_351 # First upgrade block\n",
    "END_BLOCK = 3_811_908 # NEW Fix block"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: Fill in with ss58 addresses\n",
    "# Note: this script assumes your *coldkey* was not swapped in the period between START_BLOCK and END_BLOCK\n",
    "OWNER_KEY = \"PUT OWNER SS58 ADDRESS HERE\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Assume only ever ONE hotkey swap happened\n",
    "owned_hotkeys_start = sub.query_subtensor(\"OwnedHotkeys\", START_BLOCK, params=[OWNER_KEY])\n",
    "owned_hotkeys_end = sub.query_subtensor(\"OwnedHotkeys\", END_BLOCK, params=[OWNER_KEY])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_w_pending = []\n",
    "for hk in owned_hotkeys_start:\n",
    "    pending_em = sub.query_subtensor(\"PendingdHotkeyEmission\", START_BLOCK + 10, params=[hk.value])\n",
    "    if pending_em:\n",
    "        all_w_pending.append((hk.value, pending_em.value))\n",
    "\n",
    "all_w_pending = sorted(all_w_pending, key=lambda x: x[1], reverse=True)\n",
    "START_HOTKEY = all_w_pending[0][0]\n",
    "print(f\"Start hotkey: {START_HOTKEY}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_w_pending = []\n",
    "for hk in owned_hotkeys_end:\n",
    "    pending_em = sub.query_subtensor(\"PendingdHotkeyEmission\", END_BLOCK - 120, params=[hk.value])\n",
    "    if pending_em:\n",
    "        all_w_pending.append((hk.value, pending_em.value))\n",
    "\n",
    "all_w_pending = sorted(all_w_pending, key=lambda x: x[1], reverse=True)\n",
    "END_HOTKEY = all_w_pending[0][0]\n",
    "print(f\"End hotkey: {END_HOTKEY}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "ZERO_KEY = bittensor.u8_key_to_ss58([0]*32)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    " \n",
    "TO_EMIT = {}\n",
    "EMISSIONS = {}\n",
    "\n",
    "for i, curr_hk in enumerate([START_HOTKEY, END_HOTKEY]):\n",
    "    if i == 0 and START_HOTKEY == END_HOTKEY:\n",
    "        print(\"Start and end hotkey are the same\")\n",
    "        continue\n",
    "\n",
    "    # Do for each hotkey\n",
    "    tempos = []\n",
    "    curr_block = END_BLOCK\n",
    "    while curr_block > START_BLOCK:\n",
    "        # Get last emission drain\n",
    "        last_emission_drain = sub.query_subtensor(\"LastHotkeyEmissionDrain\", curr_block, params=[curr_hk])\n",
    "        if not last_emission_drain:\n",
    "            print(f\"No last emission drain found at block {curr_block}\")\n",
    "            break\n",
    "\n",
    "        if i == 0:\n",
    "            # If the start hotkey, check for swap happened\n",
    "            zero_key_stake = sub.query_subtensor(\"Stake\", curr_block, params=[curr_hk, ZERO_KEY])\n",
    "            if zero_key_stake.value > 0:\n",
    "                print(f\"Hotkey {curr_hk} swapped before block {curr_block}\")\n",
    "                # Hotkey swapped already, skip\n",
    "            else:\n",
    "                tempos.append(last_emission_drain.value)\n",
    "                print(\"Adding tempo for start hotkey\", last_emission_drain.value)\n",
    "        else:\n",
    "            tempos.append(last_emission_drain.value)\n",
    "\n",
    "        curr_block = last_emission_drain.value - 1\n",
    "\n",
    "        \n",
    "\n",
    "    filtered_tempos = [t for t in tempos if t > 0]\n",
    "   \n",
    "    for tempo in filtered_tempos:\n",
    "        pending_emission = sub.query_subtensor(\"PendingdHotkeyEmission\", tempo - 1, params=[curr_hk])\n",
    "        if not pending_emission:\n",
    "            print(f\"No emission found at block {tempo}\")\n",
    "            continue\n",
    "\n",
    "        EMISSIONS[tempo] = pending_emission.value\n",
    "        TAKE = sub.query_subtensor(\"Delegates\", tempo - 1, params=[curr_hk])\n",
    "        if not TAKE:\n",
    "            print(f\"No delegate found at block {tempo}\")\n",
    "            raise Exception(\"No delegate found\")\n",
    "        \n",
    "        TAKE = TAKE.value / (2**16-1) # normalize to 1.0\n",
    "\n",
    "        hk_take = pending_emission.value * TAKE\n",
    "        to_emit_from_pending = pending_emission.value - hk_take\n",
    "\n",
    "        # Get the stake map\n",
    "        stake_map = sub.query_map_subtensor(\"Stake\", tempo - 1, params=[curr_hk])\n",
    "        if not stake_map:\n",
    "            print(f\"No stake map found at block {tempo}\")\n",
    "            raise Exception(\"No stake map found\")\n",
    "        \n",
    "        stake_dict = {ck.value: st.value for ck, st in stake_map}\n",
    "        \n",
    "        stake_sum = sum(stake_dict.values())\n",
    "        TO_EMIT_THIS_TEMPO = {}\n",
    "        for ck, st in stake_dict.items():\n",
    "            TO_EMIT_THIS_TEMPO[ck] = 0\n",
    "            proportion = st / stake_sum\n",
    "            owed_emission = proportion * to_emit_from_pending\n",
    "            # Track for below assertion\n",
    "            TO_EMIT_THIS_TEMPO[ck] += owed_emission\n",
    "            \n",
    "            if ck not in TO_EMIT:\n",
    "                TO_EMIT[ck] = 0\n",
    "            TO_EMIT[ck] += owed_emission\n",
    "        \n",
    "        # Verify that the sum of the emissions is less than the total to emit\n",
    "        # Using an epsilon of 1_000 RAO\n",
    "        assert sum(TO_EMIT_THIS_TEMPO.values()) <= to_emit_from_pending + 1_000\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "stake_to_distribute = sum(TO_EMIT.values())\n",
    "emissions_since = sum(EMISSIONS.values())\n",
    "owner_key_em = TO_EMIT[OWNER_KEY]\n",
    "vali_take = emissions_since - stake_to_distribute\n",
    "\n",
    "stake_to_distribute_no_owner = stake_to_distribute - owner_key_em\n",
    "\n",
    "\n",
    "print(f\"Total emissions earned: {emissions_since/1e9} TAO\")\n",
    "print(f\"Validator take: {vali_take/1e9} TAO -> {vali_take/emissions_since*100:.2f}%\")\n",
    "print(f\"Earned by owner key: {owner_key_em/1e9} TAO\")\n",
    "print(f\"Stake to distribute (minus owner key): {stake_to_distribute_no_owner/1e9} TAO\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [],
   "source": [
    "to_emit_formatted = {\n",
    "    \"address\": [],\n",
    "    \"amount\": []\n",
    "}\n",
    "total_in_csv = 0\n",
    "for coldkey, amount in TO_EMIT.items():\n",
    "    if coldkey == OWNER_KEY:\n",
    "        continue # Skip owner key\n",
    "\n",
    "    in_tao = round(amount/1e9, 9) # make into TAO, keep 9 decimal places\n",
    "    total_in_csv += in_tao\n",
    "    \n",
    "    if in_tao == 0:\n",
    "        continue # Skip coldkey if owed nothing\n",
    "    to_emit_formatted[\"address\"].append(coldkey)\n",
    "    to_emit_formatted[\"amount\"].append(in_tao) \n",
    "print(f\"Total to emit recorded in CSV: {total_in_csv} TAO\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {},
   "outputs": [],
   "source": [
    "def to_csv(emission_map: Dict[str, int], filename: str):\n",
    "    df = pd.DataFrame.from_dict(emission_map)\n",
    "    df.to_csv(filename, index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [],
   "source": [
    "to_csv(to_emit_formatted, \"emit_map.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [],
   "source": [
    "def to_json(emission_map: Dict[str, int], filename: str):\n",
    "    with open(filename, \"w\") as f:\n",
    "        json.dump(emission_map, f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [],
   "source": [
    "TO_EMIT_COPY = {k: int(round(v/1e9, 9)*1e9) for k, v in TO_EMIT.items() if k != OWNER_KEY and v > 0}\n",
    "\n",
    "to_json(TO_EMIT_COPY, \"emit_map.json\")\n",
    "print(\"Done\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "finney",
   "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.9.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}