Spaces:
Sleeping
Sleeping
| import bittensor as bt | |
| from substrateinterface import Keypair | |
| import gradio as gr | |
| import pandas as pd | |
| import time | |
| # primitive caching | |
| g_cached_data: pd.DataFrame | None = None | |
| g_last_fetch_time = 0.0 | |
| def fetch_incentive_data() -> pd.DataFrame: | |
| data = [] | |
| subtensor = bt.subtensor(network="finney") if hasattr(bt, "subtensor") else bt.Subtensor(network="finney") | |
| print("connected to subtensor") | |
| subnets = subtensor.all_subnets() | |
| print("fetched all subnets") | |
| metagraphs = subtensor.get_all_metagraphs_info() | |
| print("fetched all metagraphs") | |
| assert subnets, "WTF" | |
| assert metagraphs, "WTF" | |
| skips = [12, 20, 27, 35, 43, 53, 73, 106] | |
| for sn in range(1, 129): | |
| subnet = subnets[sn] | |
| metagraph = metagraphs[sn] | |
| hotkeys_to_uid = {hk: i for i, hk in enumerate(metagraph.hotkeys)} | |
| # The incentives that are assigned to the owner hotkey are being burned/not given out | |
| # by Maciej Kula [Bo𝞃, Bo𝞃] 23.07.2025 | |
| addresses = [("hotkey", subnet.owner_hotkey)] # So don't include ("coldkey", subnet.owner_coldkey). | |
| for key_type, address in addresses: | |
| uid = hotkeys_to_uid.get(address, None) | |
| if uid is None: | |
| continue | |
| incentive = metagraph.incentives[uid] | |
| if sn in skips: | |
| incentive = 1.0 # include these subnets regardless of burn rate | |
| # if incentive <= 1e-4: | |
| # continue | |
| is_active = metagraph.pending_root_emission.tao > 0 and metagraph.alpha_out_emission > 0 and metagraph.moving_price > 0 | |
| data.append([ | |
| f"[netuid: {sn} / {subnet.subnet_name}](https://taostats.io/subnets/{sn})" + (" 🟢" if sn not in skips else " 🔴"), | |
| is_active, | |
| round(subnet.alpha_to_tao(1).tao, 6), | |
| round(incentive*100, 2), | |
| f"[{address}](https://taostats.io/{key_type}/{address}) [{uid}]" | |
| ]) | |
| break | |
| df = pd.DataFrame(data, columns=["Subnet", "Active", "α to τ", "Burn (%)", "Address [UID]"]) # type: ignore | |
| print(f"{len(data)} subnets burn") | |
| return df | |
| def get_cached_data() -> tuple[str, pd.DataFrame]: | |
| global g_cached_data, g_last_fetch_time | |
| if g_cached_data is None or (time.time() - g_last_fetch_time) > 1200: # 20 min | |
| g_last_fetch_time = time.time() | |
| g_cached_data = fetch_incentive_data() | |
| time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(g_last_fetch_time + 1200)) | |
| return time_str, g_cached_data | |
| with gr.Blocks(title="Bittensor Subnet Incentives") as demo: | |
| next_process_text = gr.Textbox(label="Next refresh time", interactive=False) | |
| output_df = gr.DataFrame( | |
| datatype=["markdown", "bool", "number", "number", "markdown"], | |
| label="Subnet Burn Data", | |
| show_row_numbers=True, | |
| interactive=False, | |
| max_height=1000000 | |
| ) | |
| demo.load(get_cached_data, None, [next_process_text, output_df]) | |
| demo.launch(share=True) | |