app.py

"""
Streamlit dashboard for the Dippy Roleplay Subnet Leaderboard
"""
import requests
import streamlit as st
import pandas as pd
import numpy as np

st.set_page_config(layout="wide")

REMOTE_LEADERBOARD_URL = "https://dippy-bittensor-subnet.com/minerboard"

def iswin(score_i, score_j, block_i, block_j):
    MAX_PENALTY = 0.03  # Adjust this value as needed
    penalty = MAX_PENALTY
    score_i = (1 - penalty) * score_i if block_i > block_j else score_i
    score_j = (1 - penalty) * score_j if block_j > block_i else score_j
    return score_i > score_j

def calculate_win_rate(df):
    n = len(df)
    win_counts = np.zeros(n)
    
    for i in range(n):
        for j in range(n):
            if i != j:
                if iswin(df.loc[i, 'total_score'], df.loc[j, 'total_score'],
                         df.loc[i, 'block'], df.loc[j, 'block']):
                    win_counts[i] += 1
    
    return win_counts / (n - 1)  # Divide by (n-1) as each row isn't compared with itself

def apply_block_threshold_mutation(df, block_threshold):
    """
    Apply mutation to rows where block is less than the threshold.
    Changes total_score, judge_score, and notes columns for these rows.
    
    Args:
        df: DataFrame to modify
        block_threshold: Rows with block value less than this will be modified
    
    Returns:
        Modified DataFrame
    """
    # Create a copy to avoid modifying the original DataFrame
    modified_df = df.copy()
    
    # Identify rows where block is less than the threshold
    mask = modified_df['block'] < block_threshold
    
    # Apply mutations to these rows
    if mask.any():
        modified_df.loc[mask, 'total_score'] = 0.0  # Set total_score to 0
        modified_df.loc[mask, 'judge_score'] = 0.0  # Set judge_score to 0
        modified_df.loc[mask, 'notes'] = "Scoring reset"

    
    return modified_df

def leaderboard_dashboard():
    # load the logo from image.txt file as base64
    with open("image.txt", "r") as f:
        image = f.read()

    st.markdown(
        f"""
        <div style="text-align: center;">
            <img src="data:image/png;base64,{image}" alt="Dippy Roleplay Logo" width="600" height="300" style="margin-bottom: 2rem;">
            <h1 style="margin-top: 0;">SN11-Dippy-Roleplay Leaderboard</h1>
            <div style="font-size: 18px;">This is the leaderboard for the Dippy validation API hosted by SN11.</div>
        </div>
        """,
        unsafe_allow_html=True,
    )
    
    # Add emojis based on the status
    status_emojis = {
        'COMPLETED': '✅COMPLETED',
        'FAILED': '❌FAILED',
        'QUEUED': '🕒QUEUED',
        'RUNNING': '🏃RUNNING'
    }
    
    # Get the minerboard data from the API
    response = requests.get(REMOTE_LEADERBOARD_URL)
    if response.status_code != 200:
        st.error("Failed to fetch minerboard data.")
        return

    # Parse the response JSON data
    minerboard_data = response.json()

    # Convert the data to a DataFrame
    minerboard = pd.DataFrame(minerboard_data)
    
    # Set FAILED entries to have total_score as 0
    minerboard.loc[minerboard['status'] == 'FAILED', 'total_score'] = 0
    block_threshold = 5207777
    # Apply block threshold mutation
    minerboard = apply_block_threshold_mutation(minerboard, block_threshold)
    
    minerboard['status'] = minerboard['status'].map(lambda status: status_emojis.get(status, status))
    minerboard['total_score'] = minerboard['judge_score']
    
    # Sort the minerboard by the total_score column
    minerboard = minerboard.sort_values(by='total_score', ascending=False, ignore_index=True)

    front_order = ['uid', 'hotkey', 'total_score', 'status', 'chat_template_type', 'hash']
    if 'notes' in minerboard.columns:
        front_order.append('notes')

    # move status column to the front
    column_order = front_order + [column for column in minerboard.columns if column not in front_order]
    minerboard = minerboard[column_order]
    
    # Win rate calculation
    minerboard_winrate = pd.DataFrame(minerboard_data)
    minerboard_winrate['total_score'] = minerboard_winrate['judge_score']
    minerboard_winrate.loc[minerboard_winrate['status'] == 'FAILED', 'total_score'] = 0
    
    # Apply block threshold mutation to winrate dataframe too
    minerboard_winrate = apply_block_threshold_mutation(minerboard_winrate, block_threshold)
    
    minerboard_winrate['status'] = minerboard_winrate['status'].map(lambda status: status_emojis.get(status, status))
    minerboard_winrate['win_rate'] = calculate_win_rate(minerboard_winrate)
    minerboard_winrate = minerboard_winrate.sort_values(by='win_rate', ascending=False, ignore_index=True)
            
    column_order = ['uid', 'win_rate', 'hotkey', 'repo_namespace', 'repo_name', 'total_score', 'block', 'judge_score']
    if 'notes' in minerboard_winrate.columns:
        column_order.append('notes')
    
    # Create a new DataFrame with only the specified columns
    minerboard_winrate = minerboard_winrate[column_order]
    
    st.header("Leaderboard by Win Rate")
    st.dataframe(minerboard_winrate, hide_index=True)
    
    with st.expander("See detailed calculation method"):
        st.write("The win rate is calculated by comparing each miner against every other miner. Note that this board is only an approximation as queued miners have a score of 0, validators are omitted, etc.")
        st.code("""
    Example of calculating a win:
    def iswin(score_i, score_j, block_i, block_j):
        penalty = 0.03
        score_i = (1 - penalty) * score_i if block_i > block_j else score_i
        score_j = (1 - penalty) * score_j if block_j > block_i else score_j
        return score_i > score_j
    """)
    
    st.markdown("---")
    
    st.header("Minerboard")
    st.dataframe(minerboard, hide_index=True)
    
    st.markdown("---")
    
    if block_threshold > 0:
        st.info(f"Note: Entries with block values below {block_threshold} have had their scores set to 0.")


if __name__ == '__main__':
    leaderboard_dashboard()