Fix bugs

app.py

@@ -6,93 +6,99 @@ import plotly.express as px
 import utils
 
 _ = """
-Proteins folded (delta 24hr)
-Current proteins folding (24hr)
-Average time to fold trend
-Refolded proteins (group by run id and pdb id and get unique)
-Simulation duration distribution
+[x] Define KPIs: Number of steps, number of completions and total generated tokens
+[x] Data pipeline I: pull run summary data from wandb
+[x] Data pipeline II: pull run event data from wandb (max 500 steps per run)
+[x] Task trends: Number of tasks over time
+[x] Reward trends I: average reward over time, by task
+[x] Reward trends II: average nonzero reward over time, by task
+[x] Reward trends III: average nonzero normalized reward over time, by task
+[x] Explain trends: show release dates to indicate sudden changes 
+[ ] Miner trends: associate uids with miner rankings and plot top miner rewards vs network avg 
+[ ] Baseline rewards I: compare the network trends with baseline model gpt-3.5-turbo
+[ ] Baseline rewards II: compare the network trends with baseline model gpt-4o
+[ ] Baseline rewards III: compare the network trends with baseline model zephyr
+[ ] Baseline rewards IV: compare the network trends with baseline model solar
+[ ] Baseline rewards V: compare the network trends with baseline model llama3 8B
+[ ] Baseline rewards VI: compare the network trends with baseline model llama3 70B
+
+---------
 """
 
-UPDATE_INTERVAL = 3600
-
-
-st.title('Folding Subnet Dashboard')
+st.title('Prompting Subnet Dashboard')
 st.markdown('<br>', unsafe_allow_html=True)
 
 # reload data periodically
-df = utils.build_data(time.time()//UPDATE_INTERVAL)
-st.toast(f'Loaded {len(df)} runs')
-
-# TODO: fix the factor for 24 hours ago
-runs_alive_24h_ago = (df.last_event_at > pd.Timestamp.now() - pd.Timedelta('1d'))
-df_24h = df.loc[runs_alive_24h_ago]
-# correction factor to account for the fact that the data straddles the 24h boundary
-# correction factor is based on the fraction of the run which occurred in the last 24h
-# factor = (df_24h.last_event_at - pd.Timestamp.now() + pd.Timedelta('1d')) / pd.Timedelta('1d')
+state_vars = utils.load_state_vars()
 
+df_runs = state_vars['df_runs']
+df_runs_24h = state_vars['df_runs_24h']
+df_vali = state_vars['df_vali']
+df_events = state_vars['df_events']
+df_task_counts = state_vars['df_task_counts']
+df_m = state_vars['metagraph']
+st.toast(f'Loaded {len(df_runs)} runs')
 
 #### ------ PRODUCTIVITY ------
 
 # Overview of productivity
 st.subheader('Productivity overview')
-st.info('Productivity metrics show how many proteins have been folded, which is the primary goal of the subnet. Metrics are estimated using weights and biases data combined with heuristics.')
+st.info('Productivity metrics show how much data has been created by subnet 1')
 
-productivity = utils.get_productivity(df)
-productivity_24h = utils.get_productivity(df_24h)
+productivity = utils.get_productivity(df_runs)
+productivity_24h = utils.get_productivity(df_runs_24h)
 
 
-m1, m2, m3 = st.columns(3)
-m1.metric('Unique proteins folded', f'{productivity.get("unique_folded"):,.0f}', delta=f'{productivity_24h.get("unique_folded"):,.0f} (24h)')
-m2.metric('Total proteins folded', f'{productivity.get("total_simulations"):,.0f}', delta=f'{productivity_24h.get("total_simulations"):,.0f} (24h)')
-m3.metric('Total simulation steps', f'{productivity.get("total_md_steps"):,.0f}', delta=f'{productivity_24h.get("total_md_steps"):,.0f} (24h)')
+m1, m2, m3, m4 = st.columns(4)
+m1.metric('Competition duration', f'{productivity.get("duration").days} days')
+m2.metric('Total events', f'{productivity.get("total_events")/1e6:,.2f}M', delta=f'{productivity_24h.get("total_events")/1e6:,.2f}M (24h)')
+m3.metric('Total completions', f'{productivity.get("total_completions")/1e9:,.2f}B', delta=f'{productivity_24h.get("total_completions")/1e9:,.2f}B (24h)')
+m4.metric('Total dataset tokens', f'{productivity.get("total_tokens")/1e9:,.2f}B', delta=f'{productivity_24h.get("total_tokens")/1e9:,.2f}B (24h)')
 
 st.markdown('<br>', unsafe_allow_html=True)
 
-time_binned_data = df.set_index('last_event_at').groupby(pd.Grouper(freq='12h'))
-
-PROD_CHOICES = {
-    'Unique proteins folded': 'unique_pdbs',
-    'Total simulations': 'total_pdbs',
-    'Total simulation steps': 'total_md_steps',
-}
-prod_choice_label = st.radio('Select productivity metric', list(PROD_CHOICES.keys()), index=0, horizontal=True)
-prod_choice = PROD_CHOICES[prod_choice_label]
-steps_running_total = time_binned_data[prod_choice].sum().cumsum()
 st.plotly_chart(
-    # add fillgradient to make it easier to see the trend
-    px.area(steps_running_total, y=prod_choice, 
-            labels={'last_event_at':'', prod_choice: prod_choice_label},
-    ).update_traces(fill='tozeroy'),
+    px.area(df_task_counts, y=df_task_counts.columns, title='Data Created by Task', 
+            labels={'created_at':'','value':'Total data created'},
+            ),
     use_container_width=True,
 )
 
 st.markdown('<br>', unsafe_allow_html=True)
 
+# Overview of productivity
+st.subheader('Improvement overview')
+st.info('Subnet 1 is an endlessly improving system, where miners compete to produce high quality responses to a range of challenging tasks')
+
+
+TASK_CHOICES = {
+    'Question answering': 'qa',
+    'Summarization': 'summarization',
+    'Date-based question answering': 'date_qa',
+    'Math': 'math',
+    'Generic instruction': 'generic',
+    'Sentiment analysis': 'sentiment',
+    'Translation': 'translation',
+}
 
-#### ------ THROUGHPUT ------
-st.subheader('Throughput overview')
-
-st.info('Throughput metrics show the total amount of data sent and received by the validators. This is a measure of the network activity and the amount of data that is being processed by the subnet.')
-
-MEM_UNIT = 'GB' #st.radio('Select memory unit', ['TB','GB', 'MB'], index=0, horizontal=True)
+with st.expander('Advanced settings'):
+    c1, c2 = st.columns(2)
+    remove_zero_rewards = c1.checkbox('Exclude zero rewards', value=True, help='Remove completions which scored zero rewards (failed responses, timeouts etc.)')
+    normalize_rewards = c1.checkbox('Normalize rewards', value=True, help='Scale rewards for each task to a maximium value of 1 (approx)')
+    show_releases = c1.checkbox('Show releases', value=False, help='Add annotations which indicate when major releases may have impacted network performance')
+    moving_avg_window = c2.slider('Moving avg. window', min_value=1, max_value=30, value=14, help='Window size to smooth data and make long term trends clearer')
 
-data_transferred = utils.get_data_transferred(df,unit=MEM_UNIT)
-data_transferred_24h = utils.get_data_transferred(df_24h, unit=MEM_UNIT)
+reward_col = 'normalized_rewards' if normalize_rewards else 'rewards'
 
-m1, m2, m3 = st.columns(3)
-m1.metric(f'Total sent data ({MEM_UNIT})', f'{data_transferred.get("sent"):,.0f}', delta=f'{data_transferred_24h.get("sent"):,.0f} (24h)')
-m2.metric(f'Total received data ({MEM_UNIT})', f'{data_transferred.get("received"):,.0f}', delta=f'{data_transferred_24h.get("received"):,.0f} (24h)')
-m3.metric(f'Total transferred data ({MEM_UNIT})', f'{data_transferred.get("total"):,.0f}', delta=f'{data_transferred_24h.get("total"):,.0f} (24h)')
+df_stats = utils.get_reward_stats(df_events, exclude_multiturn=True, freq='1D', remove_zero_rewards=remove_zero_rewards)
 
 
-IO_CHOICES = {'total_data_sent':'Sent', 'total_data_received':'Received'}
-io_running_total = time_binned_data[list(IO_CHOICES.keys())].sum().rename(columns=IO_CHOICES).cumsum().melt(ignore_index=False)
-io_running_total['value'] = io_running_total['value'].apply(utils.convert_unit, args=(utils.BASE_UNITS, MEM_UNIT))
+task_choice_label = st.radio('Select task', list(TASK_CHOICES.keys()), index=0, horizontal=True)
+task_choice = TASK_CHOICES[task_choice_label]
 
 st.plotly_chart(
-    px.area(io_running_total, y='value', color='variable',
-            labels={'last_event_at':'', 'value': f'Data transferred ({MEM_UNIT})', 'variable':'Direction'},
-    ),
+    # add fillgradient to make it easier to see the trend
+    utils.plot_reward_trends(df_stats, task=task_choice, window=moving_avg_window, col=reward_col, annotate=show_releases, task_label=task_choice_label),
     use_container_width=True,
 )
 
@@ -107,7 +113,6 @@ m1, m2 = st.columns(2)
 ntop = m1.slider('Number of top miners to display', value=10, min_value=3, max_value=50, step=1)
 entity_choice = m2.radio('Select entity', utils.ENTITY_CHOICES, index=0, horizontal=True)
 
-df_m = utils.get_metagraph(time.time()//UPDATE_INTERVAL)
 df_miners = utils.get_leaderboard(df_m, ntop=ntop, entity_choice=entity_choice)
 
 # hide colorbar and don't show y axis
@@ -128,13 +133,13 @@ st.markdown('<br>', unsafe_allow_html=True)
 #### ------ LOGGED RUNS ------
 
 st.subheader('Logged runs')
-st.info('The timeline shows the creation and last event time of each run.')
-st.plotly_chart(
-    px.timeline(df, x_start='created_at', x_end='last_event_at', y='username', color='state',
-                labels={'created_at':'Created at', 'last_event_at':'Last event at', 'username':''},
-                ),
-    use_container_width=True
-)
+# st.info('The timeline shows the creation and last event time of each run.')
+# st.plotly_chart(
+#     px.timeline(df_runs, x_start='created_at', x_end='last_event_at', y='user', color='state',
+#                 labels={'created_at':'Created at', 'last_event_at':'Last event at', 'username':''},
+#                 ),
+#     use_container_width=True
+# )
 
 with st.expander('Show raw run data'):
-    st.dataframe(df)
+    st.dataframe(df_runs)

requirements.txt

@@ -2,4 +2,5 @@ git+https://github.com/macrocosm-os/prompting.git
 aiohttp
 deprecated
 aiohttp_apispec>=2.2.3
-aiofiles
+aiofiles
+streamlit

utils.py

@@ -1,10 +1,13 @@
 import os
 import tqdm
 import time
+import glob
 import wandb
+from traceback import print_exc
 import streamlit as st
 import pandas as pd
 import bittensor as bt
+import plotly.express as px
 
 
 # TODO: Store the runs dataframe (as in sn1 dashboard) and top up with the ones created since the last snapshot
@@ -12,14 +15,14 @@ import bittensor as bt
 
 
 MIN_STEPS = 10 # minimum number of steps in wandb run in order to be worth analyzing
-MAX_RUNS = 100#0000
 NETUID = 1
 BASE_PATH = 'macrocosmos/prompting-validators'
 NETWORK = 'finney'
-KEYS = None
+KEYS = ['_step','_timestamp','task','query','reference','challenge','topic','subtopic']
 ABBREV_CHARS = 8
 ENTITY_CHOICES = ('identity', 'hotkey', 'coldkey')
-
+LOCAL_WANDB_PATH = './data/wandb'
+USERNAME = 'opentensor'
 
 api = wandb.Api(timeout=600)
 
@@ -102,17 +105,44 @@ def get_metagraph(time):
     return df_m
 
 
-@st.cache_data()
-def load_run(run_path, keys=KEYS):
-
-    print('Loading run:', run_path)
-    run = api.run(run_path)
-    df = pd.DataFrame(list(run.scan_history(keys=keys)))
-    for col in ['updated_at', 'created_at']:
-        if col in df.columns:
-            df[col] = pd.to_datetime(df[col])
-    print(f'+ Loaded {len(df)} records')
-    return df
+@st.cache_data(show_spinner=False)
+def load_downloaded_runs(time, cols=KEYS):
+
+    list_cols = ['rewards','uids']
+    extra_cols = ['turn']
+    df_all = pd.DataFrame()
+
+    progress = st.progress(0, text='Loading downloaded data')
+    paths = glob.glob(os.path.join(LOCAL_WANDB_PATH,'*.parquet'))
+    for i, path in enumerate(paths):
+        run_id = path.split('/')[-1].split('.')[0]
+        frame = pd.read_parquet(path).dropna(subset=cols)
+        frame._timestamp = frame._timestamp.apply(pd.to_datetime, unit='s')
+        # handle missing extra cols such as turn which depend on the version of the codebase
+        found_extra_cols = [c for c in frame.columns if c in extra_cols]
+        df_long = frame[cols+list_cols+found_extra_cols].explode(list_cols)
+
+        prog_msg = f'Downloading data {i/len(paths)*100:.0f}%'
+        progress.progress(i/len(paths), text=f'{prog_msg}... **downloading** `{run_id}`')
+
+        df_all = pd.concat([df_all, df_long.assign(run_id=run_id)], ignore_index=True)
+
+    progress.empty()
+
+    # Ensure we have consistent naming schema for tasks
+    task_mapping = {
+        'date-based question answering': 'date_qa',
+        'question-answering': 'qa',
+    }
+    df_all.task = df_all.task.apply(lambda x: task_mapping.get(x, x))
+
+    # Runs which do not have a turn field are imputed to be turn zero (single turn)
+    df_all.turn.fillna(0, inplace=True)
+
+    df_all.sort_values(by=['_timestamp'], inplace=True)
+
+    return df_all
+
 
 @st.cache_data(show_spinner=False)
 def build_data(timestamp=None, path=BASE_PATH, min_steps=MIN_STEPS, use_cache=True):
@@ -140,7 +170,7 @@ def build_data(timestamp=None, path=BASE_PATH, min_steps=MIN_STEPS, use_cache=Tr
         if num_steps<min_steps:
             continue
         n_events += num_steps
-        prog_msg = f'Loading data {i/len(runs)*100:.0f}%, {n_events:,.0f} events)'
+        prog_msg = f'Loading data {i/len(runs)*100:.0f}%, (total {n_events:,.0f} events)'
         progress.progress(i/len(runs),text=f'{prog_msg}... **downloading** `{os.path.join(*run.path)}`')
 
         run_data.append(run)
@@ -153,23 +183,222 @@ def build_data(timestamp=None, path=BASE_PATH, min_steps=MIN_STEPS, use_cache=Tr
     df['identity'] = df['vali_hotkey'].map(IDENTITIES).fillna('unknown')
     df['vali_hotkey'] = df['vali_hotkey'].str[:ABBREV_CHARS]
 
+    # Drop events that are not related to validator queries
+    df.dropna(subset='query', inplace=True)
+
+    print(df.completions.apply(type).value_counts())
+    # Assumes completions is in the frame
+    df['completions'] = df['completions'].apply(lambda x: x if isinstance(x, list) else eval(x))
+
+    df['completion_words'] = df.completions.apply(lambda x: sum([len(xx.split()) for xx in x]) if isinstance(x, list) else 0)
+    df['validator_words'] = df.apply(lambda x: len(str(x.query).split()) + len(str(x.challenge).split()) + len(str(x.reference).split()), axis=1 )
+
     df.to_csv(save_path, index=False)
+
     return df
 
+@st.cache_data()
+def normalize_rewards(df, turn=0, percentile=0.98):
+    top_reward_stats = df.loc[df.turn==turn].astype({'rewards':float}).groupby('task').rewards.quantile(percentile)
+
+    df['best_reward'] = df.task.map(top_reward_stats)
+    df['normalized_rewards'] = df['rewards'].astype(float) / df['best_reward']
+    return df
+
+@st.cache_data(show_spinner=False)
+def download_runs(time, df_vali):
+
+    pbar = tqdm.tqdm(df_vali.index, total=len(df_vali))
+
+    progress = st.progress(0, text='Loading data')
+
+    for i, idx in enumerate(pbar):
+        row = df_vali.loc[idx]
+
+        prog_msg = f'Downloading data {i/len(df_vali)*100:.0f}%'
+        progress.progress(i/len(df_vali), text=f'{prog_msg}... **downloading** `{os.path.join(*row.run_id)}`')
+
+        save_path = f'data/wandb/{row.run_id}.parquet'
+        if os.path.exists(save_path):
+            pbar.set_description(f'>> Skipping {row.run_id!r} because file {save_path!r} already exists')
+            continue
+
+        try:
+            pbar.set_description(f'* Downloading run {row.run_id!r}', flush=True)
+            run = api.run(row.run_path)
+
+            # By default we just download a subset of events (500 most recent)
+            df = run.history()
+            df.to_parquet(save_path)
+        except KeyboardInterrupt:
+            break
+        except Exception as e:
+            pbar.set_description(f'- Something went wrong with {row.run_id!r}: {print_exc()}\n')
+
+    progress.empty()
+
+
+def get_productivity(df_runs):
+
+    total_duration = df_runs.last_event_at.max() - df_runs.created_at.min()
+    total_steps = df_runs.num_steps.sum()
+    total_completions = (df_runs.num_steps*df_runs.sample_size).sum()
+    total_completion_words = (df_runs.num_steps*df_runs.completion_words).sum()
+    total_completion_tokens = round(total_completion_words/0.75)
+    total_validator_words = (df_runs.num_steps*df_runs.apply(lambda x: len(str(x.query).split()) + len(str(x.challenge).split()) + len(str(x.reference).split()), axis=1 )).sum()
+    total_validator_tokens = round(total_validator_words/0.75)
+    total_dataset_tokens = total_completion_tokens + total_validator_tokens
+
+    return {
+        'duration':total_duration,
+        'total_events':total_steps,
+        'total_completions':total_completions,
+        'total_completion_tokens':total_completion_tokens,
+        'total_validator_tokens':total_validator_tokens,
+        'total_tokens':total_dataset_tokens,
+    }
+
+@st.cache_data(show_spinner=False)
+def get_reward_stats(df, exclude_multiturn=True, freq='1D', remove_zero_rewards=True, agg='mean', date_min='2024-01-22', date_max='2024-06-25'):
+
+    df = df.loc[df._timestamp.between(pd.Timestamp(date_min), pd.Timestamp(date_max))]
+    if exclude_multiturn:
+        df = df.loc[df.turn == 0]
+    if remove_zero_rewards:
+        df = df.loc[df.rewards > 0]
+
+    groups = ['run_id',pd.Grouper(key='_timestamp',freq=freq),'task']
+    return df.groupby(groups).agg({'rewards':agg, 'normalized_rewards':agg})
+
+def get_release_dates():
+    release_dates = pd.DataFrame([
+        {'version': '1.0.0', 'release_date': pd.Timestamp(month=1, day=22, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':['qa','summarization']},
+        {'version': '1.0.1', 'release_date': pd.Timestamp(month=1, day=22, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':[]},
+        {'version': '1.0.2', 'release_date': pd.Timestamp(month=1, day=24, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':['qa','summarization']},
+        {'version': '1.0.3', 'release_date': pd.Timestamp(month=2, day=14, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':[]},
+        {'version': '1.0.4', 'release_date': pd.Timestamp(month=2, day=15, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':[]},
+        {'version': '1.1.0', 'release_date': pd.Timestamp(month=2, day=21, year=2024), 'note': 'decay scores', 'model': 'zephyr', 'tasks_affected':['date_qa','math']},
+        {'version': '1.1.1', 'release_date': pd.Timestamp(month=2, day=28, year=2024), 'note': 'reduce penalty weight', 'model': 'zephyr', 'tasks_affected':['date_qa','qa','summarization']},
+        {'version': '1.1.2', 'release_date': pd.Timestamp(month=2, day=29, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':[]},
+        {'version': '1.1.3', 'release_date': pd.Timestamp(month=3, day=11, year=2024), 'note': '', 'model': 'zephyr', 'tasks_affected':[]},
+        {'version': '1.2.0', 'release_date': pd.Timestamp(month=3, day=19, year=2024), 'note': 'vllm', 'model': 'zephyr', 'tasks_affected':[]},
+        {'version': '1.3.0', 'release_date': pd.Timestamp(month=3, day=27, year=2024), 'note': '', 'model': 'solar', 'tasks_affected':['all','math']},
+        {'version': '2.0.0', 'release_date': pd.Timestamp(month=4, day=4, year=2024),  'note': 'streaming', 'model': 'solar', 'tasks_affected':['math','qa','summarization']},
+        {'version': '2.1.0', 'release_date': pd.Timestamp(month=4, day=18, year=2024), 'note': 'chattensor prompt', 'model': 'solar', 'tasks_affected':['generic']},
+        {'version': '2.2.0', 'release_date': pd.Timestamp(month=5, day=1, year=2024),  'note': 'multiturn + paraphrase', 'model': 'solar', 'tasks_affected':['sentiment','translation','math']},
+        {'version': '2.3.0', 'release_date': pd.Timestamp(month=5, day=20, year=2024), 'note': 'llama + freeform date', 'model': 'llama', 'tasks_affected':['all','date_qa']},
+        {'version': '2.3.1', 'release_date': pd.Timestamp(month=5, day=21, year=2024), 'note': '', 'model': 'llama', 'tasks_affected':['date_qa']},
+        {'version': '2.4.0', 'release_date': pd.Timestamp(month=6, day=5, year=2024),  'note': 'streaming penalty', 'model': 'llama', 'tasks_affected':[]},
+        {'version': '2.4.1', 'release_date': pd.Timestamp(month=6, day=6, year=2024),  'note': '', 'model': 'llama', 'tasks_affected':[]},
+        {'version': '2.4.2', 'release_date': pd.Timestamp(month=6, day=7, year=2024),  'note': '', 'model': 'llama', 'tasks_affected':[]},
+        {'version': '2.4.2', 'release_date': pd.Timestamp(month=6, day=7, year=2024),  'note': '', 'model': 'llama', 'tasks_affected':[]},
+        {'version': '2.5.0', 'release_date': pd.Timestamp(month=6, day=18, year=2024), 'note': 'reduce multiturn', 'model': 'llama', 'tasks_affected':['translation','sentiment']},
+        {'version': '2.5.1', 'release_date': pd.Timestamp(month=6, day=25, year=2024), 'note': 'reduce timeout', 'model': 'llama', 'tasks_affected':[]},
+    ])
+    return release_dates
+
+
+def plot_reward_trends(df_stats, task='qa', window=14, col='normalized_reward', annotate=False, task_label='Question answering'):
+
+    stats = df_stats.reset_index()
+    release_dates = get_release_dates()
+    stats_task = stats.loc[(stats.task == task)].sort_values(by='_timestamp')
+    stats_task['rewards_ma'] = stats_task[col].rolling(window, min_periods=0).mean()
+    fig = px.area(stats_task,
+            x='_timestamp', y='rewards_ma',
+            title=f'Reward Trend for {task_label} Task',
+            labels={'rewards_ma': f'Rewards [{window} day avg.]','_timestamp':''},
+            width=800,height=600,
+            )
+
+    if not annotate:
+        return fig
+
+    # Add annotations based on relevant releases
+    for idx, row in release_dates.iterrows():
+        if all(col not in row['tasks_affected'] for col in ['all',task]):
+            continue
+        # TODO add annotation or something
+        fig.add_vline(row['release_date'], line_color='red', opacity=0.6, line_dash='dot', line_width=1)#, annotation_text=str(v))
+
+    return fig
+
+@st.cache_data()
+def get_task_counts(df_runs, df_events):
+    # Get mapping from run id to prompting repo version 
+    run_to_version = df_runs.set_index('run_id').version.to_dict()
+
+    df_events['version'] = df_events.run_id.map(run_to_version)
+    
+    def version_to_spec(version):
+        major, minor, patch = version.split('.')
+        return 10_000 * major + 100 * minor + patch
+    
+    def get_closest_prev_version(version, my_versions):
+        
+        ref_spec = version_to_spec(version)
+        my_specs = list(map(version_to_spec, my_versions))
+        
+        match = my_specs[0]
+        for spec in my_specs[1:]:
+            if spec>ref_spec:
+                break
+            
+            match = spec
+        
+        return my_versions[my_specs.index(match)]
+        
+    # Now estimate the distribution of tasks for each version using the event data
+    task_rate = df_events.groupby('version').task.value_counts(normalize=True).unstack().fillna(0)
+    # Impute missing versions
+    for v in sorted(df_runs.version.unique()):
+        if v not in task_rate.index:
+            prev_version = get_closest_prev_version(v, list(task_rate.index))
+            print(f'Imputing version {v} with task rate from closes previous version {prev_version!r}')
+            task_rate.loc[v] = task_rate.loc[prev_version]
+    
+    # get esimated number of each task generated in every run using summary dataframe
+    task_counts = df_runs.set_index('created_at').sort_index().apply(lambda x: round(task_rate.loc[x.version]*x.num_steps), axis=1).cumsum()
+    return task_counts    
+
+
+def load_state_vars(username=USERNAME, percentile=0.95):
 
-def load_state_vars():
     UPDATE_INTERVAL = 600
 
-    df = build_data(time.time()//UPDATE_INTERVAL)
-    runs_alive_24h_ago = (df.last_event_at > pd.Timestamp.now() - pd.Timedelta('1d'))
-    df_24h = df.loc[runs_alive_24h_ago]
+    df_runs = build_data(time.time()//UPDATE_INTERVAL, use_cache=True)
+    
+    df_runs = df_runs.loc[df_runs.netuid.isin([1,61,102])]
+    st.toast(f'Loaded {len(df_runs)} runs')
+
+    df_vali = df_runs.loc[df_runs.username == username]
+
+    download_runs(time.time()//UPDATE_INTERVAL, df_vali)
+
+    df_events = load_downloaded_runs(time.time()//UPDATE_INTERVAL)
+    df_events = normalize_rewards(df_events, percentile=percentile)
+
+    yesterday = pd.Timestamp.now() - pd.Timedelta('1d')
+    runs_alive_24h_ago = (df_runs.last_event_at > yesterday)
+
+    df_runs_24h = df_runs.loc[runs_alive_24h_ago]
+    
+    # weight factor indicates the fraction of events that happened within the last 24 hour.
+    fraction = 1 - (yesterday - df_runs_24h.created_at) / (pd.Timestamp.now()- df_runs_24h.created_at)    
+    df_runs_24h['fraction'] = fraction.clip(0,1)
+    df_runs_24h['num_steps'] *= fraction.clip(0,1)
+    
+    df_task_counts = get_task_counts(df_runs, df_events)
 
     df_m = get_metagraph(time.time()//UPDATE_INTERVAL)
 
     return {
-        'dataframe': df,
-        'dataframe_24h': df_24h,
+        'df_runs': df_runs,
+        'df_runs_24h': df_runs_24h,
+        'df_vali': df_vali,
+        'df_events': df_events,
         'metagraph': df_m,
+        'df_task_counts': df_task_counts
     }