# Code adapted from: https://huggingface.co/spaces/RaoFoundation/pretraining-leaderboard/blob/main/app.py
import os
import datetime
from typing import Dict
import gradio as gr
from dotenv import load_dotenv
from huggingface_hub import HfApi
from apscheduler.schedulers.background import BackgroundScheduler
import utils
FONT = (
""""""
)
TITLE = """
Finetuning Subnet Leaderboard
"""
HEADER = """
Finetuning is a Bittensor subnet that rewards miners for producing finetuned models in defined competitions. The model with the best head-to-head score in each competition receive a steady emission of TAO."""
# TODO: Update links once subnet is regged.
EVALUATION_DETAILS = """
Name: the 🤗 Hugging Face model name (click to go to the model card)
Rewards / Day: the expected rewards per day based on current ranking.
Last Average Loss: the last loss value on the evaluation data for the model as calculated by a validator (lower is better)
UID: the Bittensor UID of the miner
Block: the Bittensor block that the model was submitted in
More stats on taostats."""
EVALUATION_HEADER = """
Shows the latest internal evaluation statistics as calculated by the Opentensor validator
"""
# A map of competition IDs to HTML descriptions.
COMPETITION_DETAILS: Dict[int, str] = {
1: """Competition ID 1: Produce the best fine-tuned model from a Subnet 9 pretrained model. Models are evaluated using synthetic prompt/response data from Subnet 18."""
}
HF_REPO_ID = "macrocosm-os/finetuning-leaderboard"
SECONDS_PER_BLOCK = 12
load_dotenv()
HF_TOKEN = os.environ.get("HF_TOKEN", None)
API = HfApi(token=HF_TOKEN)
def get_next_update_div(current_block: int, next_update_block: int) -> str:
now = datetime.datetime.now()
blocks_to_go = next_update_block - current_block
next_update_time = now + datetime.timedelta(
seconds=blocks_to_go * SECONDS_PER_BLOCK
)
delta = next_update_time - now
return f"""
Next reward update: {blocks_to_go} blocks (~{int(delta.total_seconds() // 60)} minutes)
Last Updated: {datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")} (UTC)
"""
def restart_space():
API.restart_space(repo_id=HF_REPO_ID, token=HF_TOKEN)
def main():
# To avoid leaderboard failures, infinitely try until we get all data
# needed to populate the dashboard
state_vars = utils.load_state_vars()
model_data = state_vars["model_data"]
vali_runs = state_vars["vali_runs"]
scores = state_vars["scores"]
validator_df = state_vars["validator_df"]
benchmarks = state_vars.get("benchmarks", None)
benchmark_timestamp = state_vars.get("benchmark_timestamp", None)
demo = gr.Blocks(css=".typewriter {font-family: 'JMH Typewriter', sans-serif;}")
with demo:
gr.HTML(FONT)
gr.HTML(TITLE)
gr.HTML(HEADER)
# TODO: Re-enable once ""SubtensorModule.BlocksSinceEpoch" not found" issue is resolved.
# gr.HTML(value=get_next_update_div(current_block, next_epoch_block))
# TODO: Figure out the best approach to showing the per competition rewards.
gr.Label(
value={
f"{c.namespace}/{c.name} ({c.commit[0:8]}) · (τ{round(c.emission, 2):,})": c.incentive
for c in model_data
if c.incentive
},
num_top_classes=10,
)
if benchmarks is not None:
with gr.Accordion("Top Model Benchmarks"):
gr.components.Dataframe(benchmarks)
gr.HTML("""
PPL computed using a stride of 512. See here for the full code.
""")
gr.HTML(f"""
Last Updated: {benchmark_timestamp.strftime("%Y-%m-%d %H:%M:%S")} (UTC)