diff --git a/flockoff/constants.py b/flockoff/constants.py index 279805d..16a9036 100644 --- a/flockoff/constants.py +++ b/flockoff/constants.py @@ -1,5 +1,4 @@ from pathlib import Path -import bittensor as bt from dataclasses import dataclass from typing import Optional @@ -8,22 +7,29 @@ MIN_WEIGHT_THRESHOLD = 1e-6 DEFAULT_RAW_SCORE = 999 DEFAULT_NORMALIZED_SCORE = 0.0 -DEFAULT_DUPLICATE_COUNT = 100 +DEFAULT_DUPLICATE_COUNT = 33 SCORE_PRECISION = 10_000 +SELECTION_SIZE = 100 # miners choose 100 rows +LOSS_THRESHOLD_PCT = 0.05 + +submission_start_utc_min = 12 * 60 +submission_window_mins = 30 +validate_start_utc_min = 12 * 60 + submission_window_mins +reward_start_utc_min = 11 * 60 + 30 @dataclass class Competition: """Class defining model parameters""" id: str = "1" - repo: str = "flock-io/flock-off-s1-character-roleplay" + repo: str = "flock-io/flock-off-s1-competition" bench: float = 2.60 minb: float = 2.40 maxb: float = 2.80 bheight: float = 0.05 pow: int = 2 - rows: int = 250 + rows: int = SELECTION_SIZE @classmethod def from_defaults(cls) -> "Competition": @@ -32,4 +38,6 @@ def from_defaults(cls) -> "Competition": # eval dataset huggingface -eval_commit = "784fbf1e78d16c512750e3bb5391fa6b338818ae" +eval_commit = "main" + + diff --git a/flockoff/miners/data.py b/flockoff/miners/data.py index 8459718..048eb07 100644 --- a/flockoff/miners/data.py +++ b/flockoff/miners/data.py @@ -8,7 +8,7 @@ # The length, in bytes, of a base64 encoded sha256 hash. SHA256_BASE_64_LENGTH = 44 # The max length, in characters, of the competition id -MAX_COMPETITION_ID_LENGTH = 2 +MAX_COMPETITION_ID_LENGTH = 12 class ModelId(BaseModel): diff --git a/flockoff/validator/database.py b/flockoff/validator/database.py index ed82640..f180a24 100644 --- a/flockoff/validator/database.py +++ b/flockoff/validator/database.py @@ -1,12 +1,14 @@ +import json import sqlite3 +import time import logging +from flockoff import constants logger = logging.getLogger(__name__) class DatabaseError(Exception): """Custom exception for database-related errors.""" - pass @@ -25,22 +27,40 @@ def _init_db(self): try: c = self.conn.cursor() c.execute( - """CREATE TABLE IF NOT EXISTS miner_scores - (uid INTEGER, - hotkey TEXT, - raw_score REAL, - normalized_score REAL, - PRIMARY KEY (uid, hotkey))""" + """CREATE TABLE IF NOT EXISTS daily_competitions ( + competition_id TEXT PRIMARY KEY, + dataset_commit TEXT, + submission_start_timestamp INTEGER, + submission_end_timestamp INTEGER, + winner_uid int, + winner_loss REAL, + status TEXT, + use_yesterday_reward INTEGER, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + );""" ) c.execute( - """CREATE TABLE IF NOT EXISTS dataset_revisions - (namespace TEXT PRIMARY KEY, revision TEXT)""" + """CREATE TABLE IF NOT EXISTS competition_submissions ( + competition_id TEXT, + uid INTEGER, + coldkey TEXT, + hotkey TEXT, + commitment_block INTEGER, + commitment_timestamp INTEGER, + eval_loss REAL, + namespace TEXT, + revision TEXT, + is_eligible INTEGER, + PRIMARY KEY (competition_id, uid), + FOREIGN KEY (competition_id) REFERENCES daily_competitions(competition_id) + )""" ) - self._add_column_if_not_exists(c, 'miner_scores', 'namespace', 'TEXT') - self._add_column_if_not_exists(c, 'miner_scores', 'revision', 'TEXT') - self._add_column_if_not_exists(c, 'miner_scores', 'raw_loss', 'REAL') + c.execute( + """CREATE TABLE IF NOT EXISTS dataset_revisions + (local_path TEXT PRIMARY KEY, namespace TEXT, revision TEXT)""" + ) self.conn.commit() except sqlite3.Error as e: logger.error(f"Failed to initialize database tables: {str(e)}") @@ -58,189 +78,173 @@ def _add_column_if_not_exists(self, cursor, table_name, column_name, column_type except sqlite3.Error as e: logger.warning(f"Failed to add column {column_name} to {table_name}: {e}") - def get_revision(self, namespace: str) -> str | None: - """Return last stored revision for this namespace (or None).""" + def get_competition_info(self, competition_id: str) -> str: + if not competition_id: + return "" try: c = self.conn.cursor() - c.execute( - "SELECT revision FROM dataset_revisions WHERE namespace = ?", - (namespace,), - ) - row = c.fetchone() + cur = c.execute("SELECT dataset_commit FROM daily_competitions WHERE competition_id = ?", (competition_id,)) + row = cur.fetchone() return row[0] if row else None except sqlite3.Error as e: - logger.error(f"Failed to get revision for namespace {namespace}: {str(e)}") - raise DatabaseError(f"Failed to retrieve revision: {str(e)}") from e + logger.error(f"Failed to get get_competition_info: {str(e)}") + raise DatabaseError(f"Failed to get get_competition_info: {str(e)}") from e - def get_score_revision(self, uid: int, namespace: str) -> str | None: - """Return last stored revision for this namespace (or None).""" + def create_competition(self, competition_id: str, submission_start_timestamp: int, dataset_commit: str): try: + status = 'submission' c = self.conn.cursor() c.execute( - "SELECT revision FROM miner_scores WHERE uid = ? AND namespace = ?", - (uid, namespace), + "INSERT INTO daily_competitions (competition_id,submission_start_timestamp,submission_end_timestamp, dataset_commit, status,use_yesterday_reward) " + "VALUES (?, ?, ?, ?, ?, 0)", + (competition_id, submission_start_timestamp, submission_start_timestamp + 60 * constants.submission_window_mins, dataset_commit, status) ) - row = c.fetchone() - return row[0] if row else None + self.conn.commit() except sqlite3.Error as e: - logger.error(f"Failed to get revision for namespace {namespace}: {str(e)}") - raise DatabaseError(f"Failed to retrieve revision: {str(e)}") from e - - def set_revision(self, namespace: str, revision: str): - """Upsert the revision for this namespace.""" + logger.error(f"Failed to create competition_id {competition_id}: {str(e)}") + raise DatabaseError(f"Failed to create competition_id: {str(e)}") from e + + def copy_competition_id(self, new_id, old_id): + sql = """ + INSERT INTO daily_competitions ( + competition_id, + dataset_commit, + submission_start_timestamp, + submission_end_timestamp, + winner_uid, + winner_loss, + status, + use_yesterday_reward + ) + SELECT + ?, + dataset_commit, + submission_start_timestamp, + submission_end_timestamp, + winner_uid, + winner_loss, + ?, + 1 + FROM daily_competitions + WHERE competition_id = ?; + """ try: c = self.conn.cursor() - c.execute( - """ - INSERT INTO dataset_revisions(namespace, revision) - VALUES (?, ?) - ON CONFLICT(namespace) DO UPDATE SET revision=excluded.revision - """, - (namespace, revision), - ) + c.execute(sql, (new_id, "rewarding", old_id)) self.conn.commit() + except sqlite3.Error as e: - logger.error(f"Failed to set revision for namespace {namespace}: {str(e)}") - raise DatabaseError(f"Failed to update revision: {str(e)}") from e + logger.error(f"Failed to copy competition_id {new_id}: {str(e)}") + raise DatabaseError(f"Failed to copy competition_id: {str(e)}") from e - def set_score_revision(self, uid: int, namespace: str, revision: str, hotkey: str): - """Upsert the revision for this namespace.""" + def update_competition_status(self, competition_id: str, status: str): try: c = self.conn.cursor() c.execute( - "UPDATE miner_scores SET namespace = ?, revision = ?, hotkey= ? WHERE uid = ?", (namespace, revision, hotkey,uid) + "UPDATE daily_competitions SET status = ? WHERE competition_id = ?", + (status, competition_id) ) self.conn.commit() except sqlite3.Error as e: - logger.error(f"Failed to set revision for namespace {namespace}: {str(e)}") - raise DatabaseError(f"Failed to update revision: {str(e)}") from e - - def insert_or_reset_uid( - self, uid: int, hotkey: str, base_raw_score: float = 0.0, initial_normalized_score: float = 1.0 / 255.0 - ): - """Insert a new UID or reset its scores if the hotkey has changed (UID recycled). - - Args: - uid (int): The UID of the miner. - hotkey (str): The hotkey of the miner. - base_raw_score (float, optional): The initial raw score to set. - Defaults to 0.0 if not provided by the caller. - initial_normalized_score (float, optional): The initial normalized score (weight). - Defaults to 1.0 / 255.0 if not provided by the caller. - """ + logger.error(f"Failed to set update_competition_status: {str(e)}") + raise DatabaseError(f"Failed to set update_competition_status: {str(e)}") from e + + def update_competition_score(self, competition_id: str, winner_uid: int, winner_loss: float): try: c = self.conn.cursor() - c.execute("SELECT hotkey FROM miner_scores WHERE uid = ?", (uid,)) - result = c.fetchone() - if result is None: - # UID doesn't exist, insert new record - c.execute( - """INSERT INTO miner_scores - (uid, hotkey, raw_score, normalized_score) - VALUES (?, ?, ?, ?)""", - (uid, hotkey, base_raw_score, initial_normalized_score), - ) - elif result[0] != hotkey: - # UID exists but hotkey changed, update the existing row with new hotkey and reset scores - c.execute( - """UPDATE miner_scores - SET hotkey = ?, raw_score = ?, normalized_score = ? - WHERE uid = ?""", - (hotkey, base_raw_score, initial_normalized_score, uid), - ) - # If UID and hotkey match, we don't reset scores, assuming they are current. - # Specific updates to raw_score or normalized_score will be handled by dedicated methods. + c.execute( + "UPDATE daily_competitions SET winner_uid=?,winner_loss=? WHERE competition_id = ?", + (winner_uid, winner_loss, competition_id) + ) self.conn.commit() except sqlite3.Error as e: - logger.error(f"Failed to insert/reset UID {uid}: {str(e)}") - raise DatabaseError(f"Failed to insert/reset UID: {str(e)}") from e + logger.error(f"Failed to set update_competition_status: {str(e)}") + raise DatabaseError(f"Failed to set update_competition_status: {str(e)}") from e - def update_raw_loss(self, uid: int, loss: float): - """Update the raw loss for a given UID.""" + def record_submission(self, competition_id: str, uid: int, hotkey: str, coldkey: str, commitment_block: int, commitment_timestamp: int,namespace: str,revision: str): try: c = self.conn.cursor() c.execute( - "UPDATE miner_scores SET raw_loss = ? WHERE uid = ?", (loss, uid) + "INSERT INTO competition_submissions (competition_id, uid, hotkey, coldkey, commitment_block, commitment_timestamp,namespace,revision) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(competition_id, uid) DO UPDATE SET hotkey = excluded.hotkey, coldkey = excluded.coldkey," + "commitment_block = excluded.commitment_block,commitment_timestamp = excluded.commitment_timestamp," + "namespace = excluded.namespace,revision = excluded.revision;", + (competition_id, uid, hotkey, coldkey, commitment_block, commitment_timestamp,namespace,revision) ) - if c.rowcount == 0: - # If somehow a UID is being updated that wasn't inserted, log a warning or error. - logger.warning(f"Attempted to update raw_loss for non-existent UID {uid}, no changes made.") self.conn.commit() except sqlite3.Error as e: - logger.error(f"Failed to update raw_loss for UID {uid}: {str(e)}") - raise DatabaseError(f"Failed to update raw_loss: {str(e)}") from e + logger.error(f"Failed to set record_submission {uid}: {str(e)}") + raise DatabaseError(f"Failed to update record_submission: {str(e)}") from e + - def update_raw_eval_score(self, uid: int, new_raw_score: float): - """Update the raw evaluation score for a given UID.""" + def record_submission_loss(self, competition_id: str, uid: int,eval_loss: float, is_eligible: bool): try: c = self.conn.cursor() c.execute( - "UPDATE miner_scores SET raw_score = ? WHERE uid = ?", (new_raw_score, uid) + "INSERT INTO competition_submissions (competition_id, uid, eval_loss, is_eligible) " + "VALUES (?, ?, ?, ?) ON CONFLICT(competition_id, uid) DO UPDATE SET eval_loss = excluded.eval_loss, is_eligible = excluded.is_eligible;", + (competition_id, uid, eval_loss, is_eligible) ) - if c.rowcount == 0: - # If somehow a UID is being updated that wasn't inserted, log a warning or error. - logger.warning(f"Attempted to update raw_score for non-existent UID {uid}, no changes made.") self.conn.commit() except sqlite3.Error as e: - logger.error(f"Failed to update raw_score for UID {uid}: {str(e)}") - raise DatabaseError(f"Failed to update raw_score: {str(e)}") from e + logger.error(f"Failed to set record_submission_loss {uid}: {str(e)}") + raise DatabaseError(f"Failed to record_submission_loss: {str(e)}") from e - def update_final_normalized_score(self, uid: int, new_normalized_score: float): - """Update the final normalized score for a given UID.""" + def get_competition_submissions(self, competition_id: str) -> dict: try: c = self.conn.cursor() - c.execute( - "UPDATE miner_scores SET normalized_score = ? WHERE uid = ?", (new_normalized_score, uid) + cur = c.execute( + "SELECT uid, coldkey, hotkey, commitment_block, commitment_timestamp, eval_loss,namespace,revision, is_eligible FROM competition_submissions WHERE competition_id = ?", + (competition_id,) ) - if c.rowcount == 0: - logger.warning(f"Attempted to update normalized_score for non-existent UID {uid}, no changes made.") - self.conn.commit() + cols = ['uid', 'coldkey', 'hotkey', 'commitment_block', 'commitment_timestamp', 'eval_loss', 'namespace','revision','is_eligible'] + rows = cur.fetchall() + return {row[0]: dict(zip(cols, row)) for row in rows} except sqlite3.Error as e: - logger.error(f"Failed to update normalized_score for UID {uid}: {str(e)}") - raise DatabaseError(f"Failed to update normalized_score: {str(e)}") from e + logger.error(f"Failed to get_competition_submissions competition_id {competition_id}: {str(e)}") + raise DatabaseError(f"Failed to get_competition_submissions: {str(e)}") from e - def get_raw_eval_score(self, uid: int) -> float | None: - """Retrieve the raw evaluation score for a given UID, defaulting to None if not found.""" + def get_competition_status(self, competition_id: str) -> str: try: c = self.conn.cursor() - c.execute("SELECT raw_score FROM miner_scores WHERE uid = ?", (uid,)) - result = c.fetchone() - return result[0] if result else None + cur = c.execute("SELECT status FROM daily_competitions WHERE competition_id = ?", (competition_id,)) + row = cur.fetchone() + return row[0] if row else None except sqlite3.Error as e: - logger.error(f"Failed to get raw_score for UID {uid}: {str(e)}") - raise DatabaseError(f"Failed to retrieve raw_score: {str(e)}") from e + logger.error(f"Failed to get get_competition_status: {str(e)}") + raise DatabaseError(f"Failed to get get_competition_status: {str(e)}") from e - def get_all_normalized_scores(self, uids: list) -> list: - """Retrieve normalized scores for a list of UIDs, defaulting to 0.0 if not found.""" - # This replaces the old get_scores for the purpose of initializing weights. + def get_revision(self, namespace: str, local_path: str) -> str: + try: + c = self.conn.cursor() + cur = c.execute("SELECT revision FROM dataset_revisions WHERE namespace = ? AND local_path = ?", (namespace, local_path)) + row = cur.fetchone() + return row[0] if row else None + except sqlite3.Error as e: + logger.error(f"Failed to get get_revision: {str(e)}") + raise DatabaseError(f"Failed to get get_revision: {str(e)}") from e + + def set_revision(self, namespace: str, revision: str, local_path: str): try: c = self.conn.cursor() - # Ensure there are UIDs to prevent empty IN clause - if not uids: - return [] - - placeholders = ','.join('?' * len(uids)) c.execute( - f"SELECT uid, normalized_score FROM miner_scores WHERE uid IN ({placeholders})", - uids, + "INSERT OR REPLACE INTO dataset_revisions (local_path, namespace, revision) VALUES (?, ?, ?)", + (local_path, namespace, revision) ) - scores_dict = {uid_val: score for uid_val, score in c.fetchall()} - # Default to 0.0 for UIDs not found in the database - return [scores_dict.get(uid_val, 0.0) for uid_val in uids] + self.conn.commit() except sqlite3.Error as e: - logger.error(f"Failed to get normalized_scores for UIDs {uids}: {str(e)}") - raise DatabaseError(f"Failed to retrieve normalized_scores: {str(e)}") from e - - def get_normalized_score(self, uid: int) -> float: - """Retrieve the normalized score for a given UID, defaulting to 0.0 if not found.""" + logger.error(f"Failed to set set_revision {local_path}: {str(e)}") + raise DatabaseError(f"Failed to record_submission_loss: {str(e)}") from e + + def get_competition_winner(self, competition_id: str) -> int: try: c = self.conn.cursor() - c.execute("SELECT normalized_score FROM miner_scores WHERE uid = ?", (uid,)) - result = c.fetchone() - return result[0] if result else 0.0 + cur = c.execute("SELECT winner_uid FROM daily_competitions WHERE competition_id = ?", (competition_id,)) + row = cur.fetchone() + return row[0] if row else None except sqlite3.Error as e: - logger.error(f"Failed to get normalized_score for UID {uid}: {str(e)}") - raise DatabaseError(f"Failed to retrieve normalized_score: {str(e)}") from e + logger.error(f"Failed to get get_competition_status: {str(e)}") + raise DatabaseError(f"Failed to get get_competition_status: {str(e)}") from e def __del__(self): """Close the connection when the instance is destroyed.""" diff --git a/flockoff/validator/trainer.py b/flockoff/validator/trainer.py index 2cb73ab..8f77beb 100644 --- a/flockoff/validator/trainer.py +++ b/flockoff/validator/trainer.py @@ -25,6 +25,7 @@ from .constants import model2template import bittensor as bt from flockoff.validator.database import ScoreDB +from flockoff import constants api = HfApi() @@ -40,7 +41,7 @@ class LoraTrainingArguments: def download_dataset( - namespace: str, revision: str, local_dir: str = "data", cache_dir: str = None, force: bool = False + namespace: str, revision: str, local_dir: str = "data", cache_dir: str = None, force: bool = False ): # Create cache directory if it doesn't exist if cache_dir: @@ -52,7 +53,7 @@ def download_dataset( local_dir = os.path.abspath(local_dir) db = ScoreDB("scores.db") - last = db.get_revision(namespace) + last = db.get_revision(namespace, local_dir) # only skip if we've recorded the same revision *and* dir still exists if last == revision and os.path.isdir(local_dir): @@ -71,12 +72,14 @@ def download_dataset( os.makedirs(local_dir, exist_ok=True) bt.logging.info(f"[HF] Downloading dataset {namespace}@{revision} → {local_dir}") - api.snapshot_download( - repo_id=namespace, local_dir=local_dir, revision=revision, repo_type="dataset" - ) - - db.set_revision(namespace, revision) - time.sleep(1) + try: + api.snapshot_download( + repo_id=namespace, local_dir=local_dir, revision=revision, repo_type="dataset" + ) + db.set_revision(namespace, revision, local_dir) + time.sleep(1) + except Exception as e: + bt.logging.error(f"api.snapshot_download error:{e}") def check_valid_revision(namespace: str, revision: str): @@ -93,6 +96,16 @@ def check_valid_revision(namespace: str, revision: str): return False return True + +def get_hg_revision(namespace: str, eval_commit: str): + try: + repo_info = HfApi(token=os.environ["HF_TOKEN"]).repo_info(repo_id=namespace, revision=eval_commit, repo_type="dataset") + except Exception as e: + bt.logging.error(f"Error fetching repo info for repo {namespace} and revision {eval_commit}: {e}") + return None + return repo_info.sha + + def reset_gpu(): """Reset GPU state and clear memory""" if torch.cuda.is_available(): @@ -109,6 +122,7 @@ def reset_gpu(): except Exception as e: bt.logging.error(f"Error resetting GPU: {e}") + def safe_cuda_cleanup(model): """Safely move model to CPU and delete it""" try: @@ -120,13 +134,14 @@ def safe_cuda_cleanup(model): finally: gc.collect() + def train_lora( - lucky_num: int, - benchmark_loss: float, - eval_size: int, - cache_dir: str = None, - data_dir: str = "data", - eval_data_dir: str = "eval_data", + lucky_num: int, + benchmark_loss: float, + eval_size: int, + cache_dir: str = None, + data_dir: str = "data", + eval_data_dir: str = "eval_data", ) -> float: try: # Reset GPU state at the start @@ -139,14 +154,7 @@ def train_lora( # set the same random seed to detect duplicate data sets from dotenv import load_dotenv - load_dotenv() - os.environ["PYTHONHASHSEED"] = str(lucky_num) - - torch.manual_seed(lucky_num) - if torch.cuda.is_available(): - torch.cuda.manual_seed(lucky_num) - torch.cuda.manual_seed_all(lucky_num) CONTEXT_LENGTH = 2048 with open(f"flockoff/validator/training_args.yaml", "r") as f: diff --git a/flockoff/validator/validator_utils.py b/flockoff/validator/validator_utils.py index b78995d..bc380a3 100644 --- a/flockoff/validator/validator_utils.py +++ b/flockoff/validator/validator_utils.py @@ -1,18 +1,21 @@ import json +from typing import Tuple, Any + import bittensor as bt import numpy as np from flockoff import constants +from flockoff.validator.database import ScoreDB def compute_score( - loss, - benchmark_loss, - min_bench, - max_bench, - power, - bench_height, - miner_comp_id, - real_comp_id, + loss, + benchmark_loss, + min_bench, + max_bench, + power, + bench_height, + miner_comp_id, + real_comp_id, ): """ Compute the score based on the loss and benchmark loss. @@ -83,6 +86,64 @@ def compute_score( denominator = np.pow((max_bench - benchmark_loss), power) return numerator / denominator + bench_height + +def select_winner(db: ScoreDB, competition_id: str, hotkeys: dict, coldkeys: dict) -> tuple[None, None] | tuple[int, float]: + subs = db.get_competition_submissions(competition_id) + scored = [s for s in subs.values() if s.get('eval_loss') is not None] + if not scored: + return None, None + + threshold_number = max(int(len(hotkeys) * constants.LOSS_THRESHOLD_PCT), 1) + scored_by_loss = sorted(scored, key=lambda s: s['eval_loss']) + eligible = scored_by_loss[:threshold_number] + + if not eligible: + return None, None + + def sort_key(s): + return (s.get('commitment_block', 10 ** 18), s.get('commitment_timestamp', 10 ** 18)) + + eligible_sorted = sorted(eligible, key=sort_key) + bt.logging.info(f"competition_id:{competition_id} , eligible_sorted:{eligible_sorted}") + winner = eligible_sorted[0] + + uid = winner['uid'] + if winner['hotkey'] != hotkeys[uid]: + replacement_found = False + for hotkey_uid, hotkey in hotkeys.items(): + if hotkey == winner['hotkey']: + winner['uid'] = hotkey_uid + replacement_found = True + if replacement_found: + bt.logging.info(f"{competition_id} competition_id winner found in hotkeys :{hotkey_uid}") + return winner['uid'], winner['eval_loss'] + + for coldkey_uid, coldkey in coldkeys.items(): + if coldkey == winner['coldkey']: + winner['uid'] = coldkey_uid + replacement_found = True + if replacement_found: + bt.logging.info(f"{competition_id} competition_id winner found in coldkeys :{coldkey_uid}") + return winner['uid'], winner['eval_loss'] + + if not replacement_found: + for candidate in eligible_sorted: + candidate_uid = candidate['uid'] + if candidate_uid != uid and hotkeys[candidate_uid] == candidate['hotkey']: + winner['uid'] = candidate_uid + bt.logging.info(f"{competition_id} competition_id winner found in eligible_sorted :{candidate_uid}") + return winner['uid'], winner['eval_loss'] + + for candidate in scored_by_loss: + candidate_uid = candidate['uid'] + if candidate_uid != uid and hotkeys[candidate_uid] == candidate['hotkey']: + winner['uid'] = candidate_uid + bt.logging.info(f"{competition_id} competition_id winner found in scored_by_loss :{candidate_uid}") + break + + return winner['uid'], winner['eval_loss'] + + def load_jsonl(path, max_rows=None): with open(path, 'r', encoding='utf-8') as f: data = [json.loads(line.strip()) for line in f if line.strip()] @@ -90,7 +151,8 @@ def load_jsonl(path, max_rows=None): data = data[:max_rows] return data + def count_similar(jsonl1, jsonl2): set1 = set(json.dumps(item, sort_keys=True) for item in jsonl1) set2 = set(json.dumps(item, sort_keys=True) for item in jsonl2) - return len(set1 & set2) \ No newline at end of file + return len(set1 & set2) diff --git a/neurons/validator.py b/neurons/validator.py index c1cf572..42b3407 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -17,34 +17,45 @@ import os import argparse import asyncio +import time + import torch import typing import random import bittensor as bt import numpy as np import json -import hashlib -from dataclasses import asdict from flockoff.constants import Competition from flockoff import constants from flockoff.utils.chain import assert_registered from flockoff.utils.git import check_and_update_code +from enum import Enum +from datetime import datetime, timezone from flockoff.validator.chain import ( retrieve_model_metadata, set_weights_with_err_msg, reveal_weights_with_err_msg, ) -from flockoff.validator.validator_utils import compute_score, load_jsonl, count_similar +from flockoff.validator.validator_utils import load_jsonl, count_similar, select_winner from flockoff.validator.trainer import ( train_lora, download_dataset, - check_valid_revision + check_valid_revision, + get_hg_revision, ) from flockoff.validator.database import ScoreDB from dotenv import load_dotenv load_dotenv() + +class CompetitionState(Enum): + SUBMISSION = "submission" + VALIDATION = "validation" + REWARDING = "rewarding" + COMPLETED = "completed" + + class Validator: @staticmethod def config(): @@ -98,6 +109,20 @@ def config(): help="Number of blocks before epoch end to set weights.", ) + parser.add_argument( + "--active_competition_id", + type=str, + default="", + help="Set the initial winner parameters.", + ) + + parser.add_argument( + "--reward_competition_id", + type=str, + default="", + help="Set the initial winner parameters.", + ) + bt.subtensor.add_args(parser) bt.logging.add_args(parser) bt.wallet.add_args(parser) @@ -149,8 +174,7 @@ def __init__(self): self.uid = assert_registered(self.wallet, self.metagraph) bt.logging.info("Initializing weights tensor") - self.weights = torch.zeros_like(torch.tensor(self.metagraph.S)) - bt.logging.info(f"Weights initialized with shape: {self.weights.shape}") + self.uids_to_eval: typing.Dict[str, typing.List] = {} bt.logging.info("Initializing score database") @@ -162,24 +186,57 @@ def __init__(self): self.last_competition_hash = None tempo = self.subtensor.tempo(self.config.netuid) self.last_submitted_epoch = ( - self.subtensor.get_next_epoch_start_block(self.config.netuid) - tempo + self.subtensor.get_next_epoch_start_block(self.config.netuid) - tempo ) self.pending_reveal: typing.Optional[dict] = None - self._update_score_init() + self.active_competition_id: str = self.config.active_competition_id + self.reward_competition_id: str = self.config.reward_competition_id + self.use_yesterday_reward: bool = False + + if self.reward_competition_id != "": + new_weights = torch.zeros_like(torch.tensor(self.metagraph.S), dtype=torch.float32) + winner = self.score_db.get_competition_winner(self.reward_competition_id) + if winner: + new_weights[winner] = 1 + self.weights = new_weights + else: + self.weights = torch.zeros_like(torch.tensor(self.metagraph.S)) + # Get the burn UID. + burn_uid = self.get_burn_uid() + self.weights[burn_uid] =1 + bt.logging.info(f"Initial weight :{self.weights}") + bt.logging.info("Validator ready to run") + def get_burn_uid(self): + # Get the subtensor owner hotkey + sn_owner_hotkey = self.subtensor.query_subtensor( + "SubnetOwnerHotkey", + params=[self.config.netuid], + ) + bt.logging.info(f"SN Owner Hotkey: {sn_owner_hotkey}") + + # Get the UID of this hotkey + sn_owner_uid = self.subtensor.get_uid_for_hotkey_on_subnet( + hotkey_ss58=sn_owner_hotkey, + netuid=self.config.netuid, + ) + bt.logging.info(f"SN Owner UID: {sn_owner_uid}") + + return sn_owner_uid + def get_registration_block(self, uid: int) -> typing.Optional[int]: """Get the block at which a UID was registered on the subnet. - + Args: uid: The unique identifier of the neuron. - + Returns: The block number when the UID was registered, or None if query fails. """ try: result = self.subtensor.query_subtensor( - "BlockAtRegistration", + "BlockAtRegistration", params=[self.config.netuid, uid] ) if result is not None: @@ -191,41 +248,6 @@ def get_registration_block(self, uid: int) -> typing.Optional[int]: bt.logging.warning(f"Failed to get registration block for UID {uid}: {e}") return None - def _update_score_init(self): - bt.logging.info("start to update score init") - current_uids = self.metagraph.uids.tolist() - competition = Competition.from_defaults() - db_normalized_scores = self.score_db.get_all_normalized_scores(current_uids) - weights = torch.tensor(db_normalized_scores, dtype=torch.float32) - - for uid in current_uids: - retrieved_raw_score = self.score_db.get_raw_eval_score(uid) - if retrieved_raw_score is not None: - normalized_score = compute_score( - retrieved_raw_score, - competition.bench, - competition.minb, - competition.maxb, - competition.pow, - competition.bheight, - competition.id, - competition.id, - ) - if uid < len(weights): - weights[uid] = normalized_score - else: - bt.logging.warning(f"UID {uid} out of bounds for new_weights tensor, skipping.") - - weights = torch.where( - weights < constants.MIN_WEIGHT_THRESHOLD, - torch.zeros_like(weights), - weights, - ) - for uid in current_uids: - if uid < len(weights): - final_normalized_weight = weights[uid].item() - self.score_db.update_final_normalized_score(uid, final_normalized_weight) - def should_set_weights(self) -> bool: current_block = self.subtensor.get_current_block() next_epoch_block = self.subtensor.get_next_epoch_start_block(self.config.netuid) @@ -247,6 +269,17 @@ async def try_sync_metagraph(self) -> bool: bt.logging.error(f"Error syncing metagraph: {e}") return False + def should_start_new_competition(self, main_commit_id: str) -> bool: + + if not self.active_competition_id: + return True + + dataset_commit_id = self.score_db.get_competition_info(self.active_competition_id) + if dataset_commit_id == main_commit_id: + # The database wasn’t updated, reward as the same as previous day. + return False + return True + async def run_step(self): bt.logging.info("Starting run step") check_and_update_code() @@ -257,205 +290,121 @@ async def run_step(self): bt.logging.warning("Failed to sync metagraph") return - bt.logging.info("Getting current UIDs and hotkeys") current_uids = self.metagraph.uids.tolist() hotkeys = self.metagraph.hotkeys - bt.logging.info(f"Current UIDs: {current_uids}") - - # Explicitly setting initial scores for new/reset UIDs. - # base_raw_score is set to constants.DEFAULT_RAW_SCORE (999), representing no prior evaluation. - base_raw_score = constants.DEFAULT_RAW_SCORE - # initial_normalized_score is set to a small non-zero value (1.0 / 255.0) - # to serve as a minimal starting weight for new miners. - initial_normalized_score = 1.0 / 255.0 - for uid in current_uids: - self.score_db.insert_or_reset_uid(uid, hotkeys[uid], base_raw_score, initial_normalized_score) - - bt.logging.info("Getting normalized scores from database for initial weights") - db_normalized_scores = self.score_db.get_all_normalized_scores(current_uids) - - bt.logging.info("Setting weights tensor from database normalized scores") - self.weights = torch.tensor(db_normalized_scores, dtype=torch.float32) - bt.logging.debug(f"Weights tensor initialized: {self.weights}") - + coldkeys = self.metagraph.coldkeys self.consensus = self.metagraph.C bt.logging.debug(f"Consensus: {self.consensus}") - is_testnet = self.config.subtensor.network == "test" - bt.logging.info(f"Network: {self.config.subtensor.network}") - bt.logging.info(f"Is testnet: {is_testnet}") - bt.logging.info("Reading chain commitment") + now = datetime.now(timezone.utc) + competition_id_today = now.strftime("%Y%m%d") + minutes_today = now.hour * 60 + now.minute competition = Competition.from_defaults() - eval_namespace = competition.repo - - bt.logging.info(f"Competition commitment: {competition}") - - bt.logging.info("Sampling competitors for evaluation") + main_commit_id = get_hg_revision(eval_namespace, constants.eval_commit) competitors = current_uids - duplicate_sample_size = min(self.config.miner_duplicate_sample_size, len(competitors)) - sample_size = min(self.config.miner_sample_size, len(competitors)) - uids_to_check_duplicate = self.rng.choice(competitors, duplicate_sample_size, replace=False).tolist() - uids_to_eval = self.rng.choice(uids_to_check_duplicate, sample_size, replace=False).tolist() - lucky_num = int.from_bytes(os.urandom(4), "little") - bt.logging.debug(f"UIDs to evaluate: {uids_to_eval}") - - raw_scores_this_epoch = {} - block_per_uid = {} - metadata_per_uid = {} # Track metadata for each UID - - duplicate_groups = [] - processed_uids = set() - bt.logging.info("Checking for duplicate scores using raw scores") - eval_data_dir = self.config.eval_data_dir - bt.logging.info( - f"Downloading eval dataset: {eval_namespace}/{constants.eval_commit}" - ) - download_dataset( - eval_namespace, - constants.eval_commit, - local_dir=eval_data_dir, - cache_dir=self.config.cache_dir, - ) - os.makedirs(eval_data_dir, exist_ok=True) - for fname in os.listdir(eval_data_dir): - if fname.endswith(".jsonl"): - src = os.path.join(eval_data_dir, fname) - dst = os.path.join(eval_data_dir, "data.jsonl") - if src != dst: - os.replace(src, dst) - bt.logging.info(f"Renamed {fname} → data.jsonl") - - for uid_i in uids_to_check_duplicate: - metadata_i = retrieve_model_metadata( - self.subtensor, self.config.netuid, self.metagraph.hotkeys[uid_i] - ) - - if metadata_i is None: - bt.logging.debug( - f"UID {uid_i} has no metadata, assigning default score" - ) - raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid_i, constants.DEFAULT_RAW_SCORE) - continue - - # Check if commitment block is greater than registration block - registration_block = self.get_registration_block(uid_i) - - metadata_per_uid[uid_i] = metadata_i # Store metadata for this UID - block_per_uid[uid_i] = metadata_i.block - bt.logging.info( - f"Downloading {uid_i}:{self.metagraph.hotkeys[uid_i]} training dataset: {metadata_i.id.namespace}/{metadata_i.id.commit}, block:{metadata_i.block}" + # set weight at the begining + if self.should_set_weights(): + bt.logging.info(f"blocks to epoch less than threshold") + bt.logging.info(f"Setting weights on chain for netuid {self.config.netuid}") + uids_py = self.metagraph.uids.tolist() + weights_py = self.weights.tolist() + # Create a fresh salt for this commitment + commit_salt = list(os.urandom(8)) + success, commit_msg, _ = set_weights_with_err_msg( + subtensor=self.subtensor, + wallet=self.wallet, + netuid=self.config.netuid, + uids=uids_py, + weights=weights_py, + wait_for_inclusion=True, + ss58_address=self.wallet.hotkey.ss58_address, + salt=commit_salt, ) - miner_i_data_dir = os.path.join(self.config.data_dir, f"miner_{uid_i}") - if registration_block is not None and metadata_i.block <= registration_block: - bt.logging.warning( - f"UID {uid_i} has commitment block {metadata_i.block} <= registration block {registration_block}. " - f"Assigning score of 0 to prevent claiming prior submissions." - ) - raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid_i, constants.DEFAULT_RAW_SCORE) - continue - download_dataset( - metadata_i.id.namespace, - metadata_i.id.commit, - local_dir=miner_i_data_dir, - cache_dir=self.config.cache_dir, - force=random.random() < 0.2 + if success: + # Persist pending reveal state using floats; helper will scale consistently + self.pending_reveal = { + "uids": uids_py, + "weights": weights_py, + "salt": commit_salt, + } + bt.logging.info("Stored pending reveal state for next interval") + else: + bt.logging.warning(f"Commit did not succeed: {commit_msg}") + next_epoch_block = self.subtensor.get_next_epoch_start_block( + self.config.netuid + ) + self.last_submitted_epoch = next_epoch_block + else: + bt.logging.info( + f"Blocks to epoch is greater than threshold, not setting weights" ) - os.makedirs(miner_i_data_dir, exist_ok=True) - - for uid_i in uids_to_check_duplicate: - if uid_i in processed_uids: - bt.logging.debug( - f"Skipping UID {uid_i} (None, zero, or already processed)" - ) - continue - - miner_i_data_dir = os.path.join(self.config.data_dir, f"miner_{uid_i}") + if self.pending_reveal is not None: try: - # Load full eval dataset for validation check - eval_data_jsonl = load_jsonl(os.path.join(eval_data_dir, "data.jsonl")) - miner_i_data_jsonl = load_jsonl(os.path.join(miner_i_data_dir, "data.jsonl"), max_rows=competition.rows) - except FileNotFoundError as e: - bt.logging.warning(f"Data file not found for UID {uid_i}: {e}") - bt.logging.info(f"Assigning fallback score to UID {uid_i} due to missing data file") - raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid_i, constants.DEFAULT_RAW_SCORE) - continue - except Exception as e: - bt.logging.error(f"Error loading data files for UID {uid_i}: {e}") - bt.logging.info(f"Assigning fallback score to UID {uid_i} due to data loading error") - raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid_i, constants.DEFAULT_RAW_SCORE) - continue - - if count_similar(eval_data_jsonl, miner_i_data_jsonl) != len(miner_i_data_jsonl): - raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid_i, constants.DEFAULT_RAW_SCORE) - bt.logging.info( - f"Assigned fallback score {constants.DEFAULT_RAW_SCORE:.6f} to UID {uid_i} due to the " - f"miner dataset is not entirely from the evaluation dataset" + bt.logging.info("Attempting to reveal previously committed weights") + reveal_success, reveal_msg, _ = reveal_weights_with_err_msg( + subtensor=self.subtensor, + wallet=self.wallet, + netuid=self.config.netuid, + uids=self.pending_reveal["uids"], + weights=self.pending_reveal["weights"], + salt=self.pending_reveal["salt"], + wait_for_inclusion=True, ) - continue - - for uid_j in uids_to_eval: - if ( - uid_i != uid_j - and uid_j not in processed_uids - ): - similar_uids = [uid_i] - miner_j_data_dir = os.path.join(self.config.data_dir, f"miner_{uid_j}") - metadata_j = metadata_per_uid.get(uid_j) - if metadata_j is None: - bt.logging.debug( - f"Skipping UID {uid_j} (metadata is None)" - ) - continue - try: - os.makedirs(miner_j_data_dir, exist_ok=True) - download_dataset( - metadata_j.id.namespace, - metadata_j.id.commit, - local_dir=miner_j_data_dir, - cache_dir=self.config.cache_dir, + if reveal_success: + bt.logging.info(f"Reveal succeeded: {reveal_msg}") + self.pending_reveal = None + else: + bt.logging.info(f"Reveal not successful yet: {reveal_msg}") + except Exception as e: + bt.logging.error(f"Reveal attempt failed: {e}") + + # SUBMISSION + if constants.submission_start_utc_min <= minutes_today < constants.validate_start_utc_min: + # record exists, the task is already created + if self.score_db.get_competition_info(competition_id_today): + if self.use_yesterday_reward: + time.sleep(10) + return + else: + for uid in competitors: + metadata = retrieve_model_metadata( + self.subtensor, self.config.netuid, self.metagraph.hotkeys[uid] ) + if metadata is not None: + self.score_db.record_submission(self.active_competition_id, + uid, hotkeys[uid], coldkeys[uid], metadata.block, + int(time.time()), metadata.id.namespace, + metadata.id.commit) + return + + if self.should_start_new_competition(main_commit_id): + bt.logging.info("STARTING NEW COMPETITION CYCLE") + self.score_db.update_competition_status(self.active_competition_id, CompetitionState.COMPLETED.value) + self.active_competition_id = competition_id_today + self.use_yesterday_reward = False + self.score_db.create_competition(self.active_competition_id, int(now.timestamp()), main_commit_id) - miner_j_data_jsonl = load_jsonl(os.path.join(miner_j_data_dir, "data.jsonl"), max_rows=competition.rows) - except FileNotFoundError as e: - bt.logging.warning(f"Data file not found for UID {uid_j} during duplicate check: {e}") - continue - except Exception as e: - bt.logging.error(f"Error loading data file for UID {uid_j} during duplicate check: {e}") - continue + else: + bt.logging.info("COPY COMPETITION REWARD BEFORE") + bt.logging.info(f"weights set by reward_competition_id {self.reward_competition_id}") + self.use_yesterday_reward = True + self.score_db.copy_competition_id(competition_id_today, self.active_competition_id) + self.score_db.update_competition_status(self.active_competition_id, CompetitionState.COMPLETED.value) + self.active_competition_id = competition_id_today + + # VALIDATION + elif constants.validate_start_utc_min <= minutes_today < 24 * 60 or \ + minutes_today < constants.reward_start_utc_min: + + if self.use_yesterday_reward or self.active_competition_id == "": + time.sleep(10) + bt.logging.info(f"now the weight is {self.weights}") + return - if count_similar(miner_j_data_jsonl, miner_i_data_jsonl) > constants.DEFAULT_DUPLICATE_COUNT: - bt.logging.debug( - f"Found similar raw score: {uid_i} and {uid_j}" - ) - similar_uids.append(uid_j) - - if len(similar_uids) > 1: - bt.logging.info(f"Found duplicate group: {similar_uids}") - duplicate_groups.append(similar_uids) - processed_uids.update(similar_uids) - - duplicates = set() - for group in duplicate_groups: - bt.logging.info(f"Processing duplicate group: {group}") - group.sort(key=lambda uid: block_per_uid[uid]) - bt.logging.info(f"Sorted by block: {group}") - - for uid in group[1:]: - duplicates.add(uid) - raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid, constants.DEFAULT_RAW_SCORE) - - for uid in uids_to_eval: - # Try to reveal previously committed weights if any if self.pending_reveal is not None: try: bt.logging.info("Attempting to reveal previously committed weights") @@ -469,200 +418,281 @@ async def run_step(self): wait_for_inclusion=True, ) if reveal_success: - bt.logging.success(f"Reveal succeeded: {reveal_msg}") + bt.logging.info(f"Reveal succeeded: {reveal_msg}") self.pending_reveal = None else: bt.logging.info(f"Reveal not successful yet: {reveal_msg}") except Exception as e: bt.logging.error(f"Reveal attempt failed: {e}") - current_raw_score = raw_scores_this_epoch.get(uid) - if current_raw_score is not None and current_raw_score == constants.DEFAULT_RAW_SCORE: - bt.logging.info(f"The dataset for UID {uid} is invalid.") - continue - bt.logging.info(f"Evaluating UID: {uid}") + # set status + self.score_db.update_competition_status(self.active_competition_id, CompetitionState.VALIDATION.value) + + duplicate_sample_size = min(self.config.miner_duplicate_sample_size, len(competitors)) + sample_size = min(self.config.miner_sample_size, len(competitors)) + uids_to_check_duplicate = self.rng.choice(competitors, duplicate_sample_size, replace=False).tolist() + uids_to_eval = self.rng.choice(uids_to_check_duplicate, sample_size, replace=False).tolist() + lucky_num = int.from_bytes(os.urandom(4), "little") + bt.logging.debug(f"UIDs to evaluate: {uids_to_eval}") + + raw_scores_this_epoch = {} + block_per_uid = {} + + duplicate_groups = [] + processed_uids = set() + bt.logging.info("Checking for duplicate scores using raw scores") + eval_data_dir = self.config.eval_data_dir bt.logging.info( - f"Retrieving model metadata for hotkey: {self.metagraph.hotkeys[uid]}" + f"Downloading eval dataset: {eval_namespace}/{main_commit_id}" + ) + download_dataset( + eval_namespace, + main_commit_id, + local_dir=eval_data_dir, + cache_dir=self.config.cache_dir, ) - metadata = metadata_per_uid.get(uid) + os.makedirs(eval_data_dir, exist_ok=True) + for fname in os.listdir(eval_data_dir): + if fname.endswith(".jsonl"): + src = os.path.join(eval_data_dir, fname) + dst = os.path.join(eval_data_dir, "data.jsonl") + if src != dst: + os.replace(src, dst) + bt.logging.info(f"Renamed {fname} → data.jsonl") + + metadata_competition_all = self.score_db.get_competition_submissions(self.active_competition_id) + for uid_i in uids_to_check_duplicate: + + if uid_i not in metadata_competition_all: + bt.logging.debug( + f"UID {uid_i} has no metadata, assigning default score" + ) + raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid_i, constants.DEFAULT_RAW_SCORE, is_eligible=False) + continue + + block_per_uid[uid_i] = metadata_competition_all[uid_i]["commitment_block"] + metadata_i_namespace = metadata_competition_all[uid_i]["namespace"] + metadata_i_commit = metadata_competition_all[uid_i]["revision"] - if self.should_set_weights(): bt.logging.info( - f"approaching weight setting time for netuid {self.config.netuid}, breaking from eval loop" + f"Downloading {uid_i}:{self.metagraph.hotkeys[uid_i]} training dataset: {metadata_i_namespace}/{metadata_i_commit}, block:{block_per_uid[uid_i]}" ) - break - - if metadata is not None: - bt.logging.info(f"Retrieved metadata: {metadata}") - ns = metadata.id.namespace - revision = metadata.id.commit - last_rev = self.score_db.get_score_revision(uid, ns) - bt.logging.info(f"Metadata namespace: {ns}, commit: {revision}") - if not check_valid_revision(namespace=ns, revision=revision): - raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid, constants.DEFAULT_RAW_SCORE) - bt.logging.info( - f"Assigned fallback score {constants.DEFAULT_RAW_SCORE:.6f} to UID {uid} due to the dataset hash is invalid" + miner_i_data_dir = os.path.join(self.config.data_dir, f"miner_{uid_i}") + + download_dataset( + metadata_i_namespace, + metadata_i_commit, + local_dir=miner_i_data_dir, + cache_dir=self.config.cache_dir, + force=random.random() < 0.2 + ) + os.makedirs(miner_i_data_dir, exist_ok=True) + + for uid_i in uids_to_check_duplicate: + if uid_i in processed_uids: + bt.logging.debug( + f"Skipping UID {uid_i} (None, zero, or already processed)" ) continue - if last_rev == revision: + + miner_i_data_dir = os.path.join(self.config.data_dir, f"miner_{uid_i}") + + try: + # Load full eval dataset for validation check + eval_data_jsonl = load_jsonl(os.path.join(eval_data_dir, "data.jsonl")) + miner_i_data_jsonl = load_jsonl(os.path.join(miner_i_data_dir, "data.jsonl"), max_rows=competition.rows) + except FileNotFoundError as e: + bt.logging.warning(f"Data file not found for UID {uid_i}: {e}") + bt.logging.info(f"Assigning fallback score to UID {uid_i} due to missing data file") + raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid_i, constants.DEFAULT_RAW_SCORE, is_eligible=False) + continue + except Exception as e: + bt.logging.error(f"Error loading data files for UID {uid_i}: {e}") + bt.logging.info(f"Assigning fallback score to UID {uid_i} due to data loading error") + raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid_i, constants.DEFAULT_RAW_SCORE, is_eligible=False) + continue + + if count_similar(eval_data_jsonl, miner_i_data_jsonl) != len(miner_i_data_jsonl): + raw_scores_this_epoch[uid_i] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid_i, constants.DEFAULT_RAW_SCORE, is_eligible=False) bt.logging.info( - f"Skipping UID {uid} as it has already been evaluated with revision {revision}" + f"Assigned fallback score {constants.DEFAULT_RAW_SCORE:.6f} to UID {uid_i} due to the " + f"miner dataset is not entirely from the evaluation dataset" ) - retrieved_raw_score = self.score_db.get_raw_eval_score(uid) - raw_scores_this_epoch[uid] = retrieved_raw_score if retrieved_raw_score is not None else constants.DEFAULT_RAW_SCORE continue - try: - miner_data_dir = os.path.join(self.config.data_dir, f"miner_{uid}") - eval_data_dir = self.config.eval_data_dir - - bt.logging.info(f"Using data directory: {miner_data_dir}") - bt.logging.info(f"Using evaluation directory: {eval_data_dir}") - - for fname in os.listdir(eval_data_dir): - if fname.endswith(".jsonl"): - src = os.path.join(eval_data_dir, fname) - dst = os.path.join(eval_data_dir, "data.jsonl") - if src != dst: - os.replace(src, dst) - bt.logging.info(f"Renamed {fname} → data.jsonl") - - bt.logging.info("Starting LoRA training") - eval_loss = train_lora( - lucky_num, - competition.bench, - competition.rows, - cache_dir=self.config.cache_dir, - data_dir=miner_data_dir, - eval_data_dir=eval_data_dir, - ) - bt.logging.info(f"Training complete with eval loss: {eval_loss}") - raw_scores_this_epoch[uid] = eval_loss - self.score_db.update_raw_eval_score(uid, eval_loss) - self.score_db.update_raw_loss(uid, eval_loss) - self.score_db.set_score_revision(uid, ns, revision, self.metagraph.hotkeys[uid]) + for uid_j in uids_to_eval: + if ( + uid_i != uid_j + and uid_j not in processed_uids + ): + similar_uids = [uid_i] + miner_j_data_dir = os.path.join(self.config.data_dir, f"miner_{uid_j}") + + if uid_j not in metadata_competition_all: + bt.logging.debug( + f"Skipping UID {uid_j} (metadata is None)" + ) + continue + try: + metadata_j_namespace = metadata_competition_all[uid_j]["namespace"] + metadata_j_commit = metadata_competition_all[uid_j]["revision"] + os.makedirs(miner_j_data_dir, exist_ok=True) + download_dataset( + metadata_j_namespace, + metadata_j_commit, + local_dir=miner_j_data_dir, + cache_dir=self.config.cache_dir, + ) + + miner_j_data_jsonl = load_jsonl(os.path.join(miner_j_data_dir, "data.jsonl"), max_rows=competition.rows) + except FileNotFoundError as e: + bt.logging.warning(f"Data file not found for UID {uid_j} during duplicate check: {e}") + continue + except Exception as e: + bt.logging.error(f"Error loading data file for UID {uid_j} during duplicate check: {e}") + continue + + if count_similar(miner_j_data_jsonl, miner_i_data_jsonl) > constants.DEFAULT_DUPLICATE_COUNT: + bt.logging.debug( + f"Found similar raw score: {uid_i} and {uid_j}" + ) + similar_uids.append(uid_j) + + if len(similar_uids) > 1: + bt.logging.info(f"Found duplicate group: {similar_uids}") + duplicate_groups.append(similar_uids) + processed_uids.update(similar_uids) + + duplicates = set() + for group in duplicate_groups: + bt.logging.info(f"Processing duplicate group: {group}") + group.sort(key=lambda uid: block_per_uid[uid]) + bt.logging.info(f"Sorted by block: {group}") + + for uid in group[1:]: + duplicates.add(uid) + raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid, constants.DEFAULT_RAW_SCORE, is_eligible=False) - bt.logging.info(f"Stored evaluation results for UID {uid}") + for uid in uids_to_eval: - except Exception as e: - bt.logging.error(f"train error: {e}") - if "CUDA" in str(e): - bt.logging.error("CUDA error detected, terminating process") - os._exit(1) - raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid, constants.DEFAULT_RAW_SCORE) - bt.logging.info( - f"Assigned fallback score {constants.DEFAULT_RAW_SCORE:.6f} to UID {uid} due to train error" - ) - else: - bt.logging.warning(f"No metadata found for UID {uid}") - raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE - self.score_db.update_raw_eval_score(uid, constants.DEFAULT_RAW_SCORE) - - bt.logging.info("Normalizing raw scores") - normalized_scores_this_epoch = {} - for uid in uids_to_check_duplicate: - current_raw_score = raw_scores_this_epoch.get(uid) - if current_raw_score is not None: - bt.logging.debug( - f"Computing normalized score for UID {uid} with raw score {current_raw_score}" + current_raw_score = raw_scores_this_epoch.get(uid) + if current_raw_score is not None and current_raw_score == constants.DEFAULT_RAW_SCORE: + bt.logging.info(f"The dataset for UID {uid} is invalid.") + continue + bt.logging.info(f"Evaluating UID: {uid}") + bt.logging.info( + f"Retrieving model metadata for hotkey: {self.metagraph.hotkeys[uid]}" ) - if competition.bench is None or competition.bench <= 0: - bt.logging.warning( - f"Invalid benchmark ({competition.bench}) for UID {uid}; defaulting score to 0" - ) - normalized_score = constants.DEFAULT_NORMALIZED_SCORE - else: - # Get the metadata for this specific UID - uid_metadata = metadata_per_uid.get(uid) - if uid_metadata is not None and uid_metadata.id is not None: - competition_id = uid_metadata.id.competition_id - else: - bt.logging.warning(f"No metadata found for UID {uid} during normalization, using None for competition_id") - competition_id = None - - normalized_score = compute_score( - current_raw_score, - competition.bench, - competition.minb, - competition.maxb, - competition.pow, - competition.bheight, - competition_id, - competition.id, - ) - normalized_scores_this_epoch[uid] = normalized_score - else: - # It's possibly due to the should_set_weights function causing data loss - bt.logging.debug(f"Save the original score for UID {uid} as raw score was missing") - # normalized_scores_this_epoch[uid] = self.weights[uid] - bt.logging.debug(f"Normalized scores for this epoch: {normalized_scores_this_epoch}") - - bt.logging.info("Creating new weights tensor based on this epoch's normalized scores") - new_weights = self.weights.clone() - for uid, norm_score in normalized_scores_this_epoch.items(): - if uid < len(new_weights): - new_weights[uid] = norm_score - else: - bt.logging.warning(f"UID {uid} out of bounds for new_weights tensor, skipping.") - new_weights = torch.where( - new_weights < constants.MIN_WEIGHT_THRESHOLD, - torch.zeros_like(new_weights), - new_weights, - ) - bt.logging.debug( - f"Thresholded new_weights (min {constants.MIN_WEIGHT_THRESHOLD}): {new_weights}" - ) + if self.should_set_weights(): + bt.logging.info( + f"approaching weight setting time for netuid {self.config.netuid}, breaking from eval loop" + ) + break + + if uid in metadata_competition_all: + ns = metadata_competition_all[uid]["namespace"] + revision = metadata_competition_all[uid]["revision"] + bt.logging.info(f"Metadata namespace: {ns}, commit: {revision}") + + if not check_valid_revision(namespace=ns, revision=revision): + raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid, constants.DEFAULT_RAW_SCORE, is_eligible=False) + bt.logging.info( + f"Assigned fallback score {constants.DEFAULT_RAW_SCORE:.6f} to UID {uid} due to the dataset hash is invalid" + ) + continue + if metadata_competition_all[uid]["eval_loss"] is not None: + bt.logging.info( + f"Skipping UID {uid} as it has already been evaluated with revision {revision}" + ) + raw_scores_this_epoch[uid] = metadata_competition_all[uid]["eval_loss"] + continue + try: + miner_data_dir = os.path.join(self.config.data_dir, f"miner_{uid}") + eval_data_dir = self.config.eval_data_dir + + bt.logging.info(f"Using data directory: {miner_data_dir}") + bt.logging.info(f"Using evaluation directory: {eval_data_dir}") + + for fname in os.listdir(eval_data_dir): + if fname.endswith(".jsonl"): + src = os.path.join(eval_data_dir, fname) + dst = os.path.join(eval_data_dir, "data.jsonl") + if src != dst: + os.replace(src, dst) + bt.logging.info(f"Renamed {fname} → data.jsonl") + + bt.logging.info("Starting LoRA training") + eval_loss = train_lora( + lucky_num, + competition.bench, + competition.rows, + cache_dir=self.config.cache_dir, + data_dir=miner_data_dir, + eval_data_dir=eval_data_dir, + ) + bt.logging.info(f"Training complete with eval loss: {eval_loss}") - bt.logging.info("Updating database with final normalized scores (weights)") - for uid in uids_to_eval: - if uid < len(new_weights): - final_normalized_weight = new_weights[uid].item() - self.score_db.update_final_normalized_score(uid, final_normalized_weight) + raw_scores_this_epoch[uid] = eval_loss + self.score_db.record_submission_loss(self.active_competition_id, uid, eval_loss, is_eligible=True) - self.weights = new_weights - bt.logging.debug(f"Updated self.weights: {self.weights}") - bt.logging.debug(f"Consensus: {self.consensus}") + bt.logging.info(f"Stored evaluation results for UID {uid}") - bt.logging.info("Setting weights on chain") - uids_py = self.metagraph.uids.tolist() - weights_py = new_weights.tolist() + except Exception as e: + bt.logging.error(f"train error: {e}") + if "CUDA" in str(e): + bt.logging.error("CUDA error detected, terminating process") + os._exit(1) + raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid, constants.DEFAULT_RAW_SCORE, is_eligible=False) + bt.logging.info( + f"Assigned fallback score {constants.DEFAULT_RAW_SCORE:.6f} to UID {uid} due to train error" + ) + else: + bt.logging.warning(f"No metadata found for UID {uid}") + raw_scores_this_epoch[uid] = constants.DEFAULT_RAW_SCORE + self.score_db.record_submission_loss(self.active_competition_id, uid, constants.DEFAULT_RAW_SCORE, is_eligible=False) + + # REWARDING + elif constants.reward_start_utc_min <= minutes_today < constants.submission_start_utc_min: + + if self.use_yesterday_reward or self.active_competition_id == "": + time.sleep(10) + bt.logging.info(f"now the weight is {self.weights}") + return + # set status + if self.score_db.get_competition_status(self.active_competition_id) == CompetitionState.REWARDING.value: + time.sleep(10) + return + winner, winner_loss = select_winner(self.score_db, self.active_competition_id, self.metagraph.hotkeys, self.metagraph.coldkeys) + bt.logging.info(f"Competition_id {self.active_competition_id} winners is {winner} ") + if winner: + new_weights = torch.zeros_like(torch.tensor(self.metagraph.S), dtype=torch.float32) + + if winner < len(new_weights): + new_weights[winner] = 1 + else: + bt.logging.warning(f"UID {winner} out of bounds for new_weights tensor, skipping.") + self.weights = new_weights + self.reward_competition_id = self.active_competition_id + self.score_db.update_competition_status(self.active_competition_id, CompetitionState.REWARDING.value) + self.score_db.update_competition_score(self.active_competition_id, winner, winner_loss) + bt.logging.info(f"weights set by reward_competition_id {self.reward_competition_id}") - if self.should_set_weights(): - bt.logging.info(f"blocks to epoch less than threshold") - bt.logging.info(f"Setting weights on chain for netuid {self.config.netuid}") - # Create a fresh salt for this commitment - commit_salt = list(os.urandom(8)) - success, commit_msg, _ = set_weights_with_err_msg( - subtensor=self.subtensor, - wallet=self.wallet, - netuid=self.config.netuid, - uids=uids_py, - weights=weights_py, - wait_for_inclusion=True, - ss58_address=self.wallet.hotkey.ss58_address, - salt=commit_salt, - ) - if success: - # Persist pending reveal state using floats; helper will scale consistently - self.pending_reveal = { - "uids": uids_py, - "weights": weights_py, - "salt": commit_salt, - } - bt.logging.info("Stored pending reveal state for next interval") else: - bt.logging.warning(f"Commit did not succeed: {commit_msg}") - next_epoch_block = self.subtensor.get_next_epoch_start_block( - self.config.netuid - ) - self.last_submitted_epoch = next_epoch_block + bt.logging.error(f"There is no score for Competition_id {self.active_competition_id}") + raise RuntimeError(f"No winners for competition_id {self.active_competition_id}") + else: - bt.logging.info( - f"Blocks to epoch is greater than threshold, not setting weights" - ) + bt.logging.error(f"The minutes time is error: {minutes_today}") + raise RuntimeError(f"No winners for competition_id {self.active_competition_id}") + async def run(self): while True: @@ -670,4 +700,4 @@ async def run(self): if __name__ == "__main__": - asyncio.run(Validator().run()) \ No newline at end of file + asyncio.run(Validator().run()) diff --git a/tests/FlockDataset/validators/test_database.py b/tests/FlockDataset/validators/test_database.py index fb199ab..3fcffbb 100644 --- a/tests/FlockDataset/validators/test_database.py +++ b/tests/FlockDataset/validators/test_database.py @@ -10,144 +10,174 @@ def db(): def test_init_db(db): - """Test that the database initializes with the correct table and columns.""" + """Test that the database initializes with the correct tables.""" c = db.conn.cursor() - c.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name='miner_scores'" + + # Check for daily_competitions + c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='daily_competitions'") + assert c.fetchone() is not None, "Table 'daily_competitions' should exist" + + # Check for competition_submissions + c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='competition_submissions'") + assert c.fetchone() is not None, "Table 'competition_submissions' should exist" + + # Check for dataset_revisions + c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='dataset_revisions'") + assert c.fetchone() is not None, "Table 'dataset_revisions' should exist" + + +def test_competition_lifecycle(db): + """Test creating a competition, retrieving info, and updating status.""" + comp_id = "comp_2023_10_01" + start_time = 1700000000 + commit_hash = "abc12345" + + # 1. Create Competition + db.create_competition(comp_id, start_time, commit_hash) + + # 2. Get Info + retrieved_commit = db.get_competition_info(comp_id) + assert retrieved_commit == commit_hash, "Should retrieve the correct dataset commit" + + # 3. Check Initial Status + status = db.get_competition_status(comp_id) + assert status == "submission", "Initial status should be 'submission'" + + # 4. Update Status + db.update_competition_status(comp_id, "evaluating") + new_status = db.get_competition_status(comp_id) + assert new_status == "evaluating", "Status should be updated to 'evaluating'" + + +def test_record_submission(db): + """Test recording and retrieving miner submissions.""" + comp_id = "comp_1" + db.create_competition(comp_id, 1000, "commit_x") + + uid = 5 + hotkey = "hk_5" + coldkey = "ck_5" + + # 1. Record Submission + db.record_submission( + competition_id=comp_id, + uid=uid, + hotkey=hotkey, + coldkey=coldkey, + commitment_block=100, + commitment_timestamp=1001, + namespace="hf_repo", + revision="main" + ) + + # 2. Retrieve Submissions + submissions = db.get_competition_submissions(comp_id) + assert uid in submissions + data = submissions[uid] + assert data['hotkey'] == hotkey + assert data['namespace'] == "hf_repo" + assert data['eval_loss'] is None # Should be None initially + + # 3. Update Submission (Upsert) - Change revision + db.record_submission( + competition_id=comp_id, + uid=uid, + hotkey=hotkey, + coldkey=coldkey, + commitment_block=105, + commitment_timestamp=1005, + namespace="hf_repo", + revision="dev_branch" ) - assert c.fetchone() is not None, "The 'miner_scores' table should be created" - c.execute("PRAGMA table_info(miner_scores)") - columns = {row[1] for row in c.fetchall()} - expected_columns = {"uid", "hotkey", "raw_score", "normalized_score", "namespace", "revision", "raw_loss"} - assert columns == expected_columns, f"Expected columns {expected_columns}, got {columns}" - - -def test_insert_or_reset_uid(db): - """Test inserting or resetting UIDs with hotkeys, raw_score, and normalized_score.""" - base_raw = 0.05 # Example base raw score - initial_norm = 0.0 # Example initial normalized score - - # Insert new UID - db.insert_or_reset_uid(1, "hotkey1", base_raw, initial_norm) - raw_score = db.get_raw_eval_score(1) - norm_score = db.get_normalized_score(1) - assert raw_score == pytest.approx(base_raw), f"New UID should have raw_score {base_raw}" - assert norm_score == pytest.approx(initial_norm), f"New UID should have normalized_score {initial_norm}" - - # Update scores and check no reset with same hotkey - db.update_raw_eval_score(1, 0.75) - db.update_final_normalized_score(1, 0.85) - raw_score = db.get_raw_eval_score(1) - norm_score = db.get_normalized_score(1) - assert raw_score == 0.75, "Raw score should be updated" - assert norm_score == 0.85, "Normalized score should be updated" - - db.insert_or_reset_uid(1, "hotkey1", base_raw, initial_norm) # This should not overwrite existing if hotkey matches - raw_score_after_reinsert = db.get_raw_eval_score(1) - norm_score_after_reinsert = db.get_normalized_score(1) - assert raw_score_after_reinsert == 0.75, "Raw score should remain after re-insert with same hotkey" - assert norm_score_after_reinsert == 0.85, "Normalized score should remain after re-insert with same hotkey" - - # Reset scores with different hotkey - new_base_raw = 0.06 - new_initial_norm = 0.01 - db.insert_or_reset_uid(1, "hotkey2", new_base_raw, new_initial_norm) - raw_score = db.get_raw_eval_score(1) - norm_score = db.get_normalized_score(1) - assert raw_score == pytest.approx(new_base_raw), f"Raw score should reset to {new_base_raw} with new hotkey" - assert norm_score == pytest.approx(new_initial_norm), f"Normalized score should reset to {new_initial_norm} with new hotkey" - - -def test_update_raw_eval_score(db): - """Test updating raw evaluation scores for UIDs.""" - db.insert_or_reset_uid(1, "hotkey1", 0.1, 0.0) - db.update_raw_eval_score(1, 0.77) - raw_score = db.get_raw_eval_score(1) - assert raw_score == 0.77, "Raw score should be updated to 0.77" - - # Test updating non-existent UID (should not raise error, no change) - db.update_raw_eval_score(2, 0.88) - raw_score_2 = db.get_raw_eval_score(2) - assert raw_score_2 is None, "Raw score for non-existent UID 2 should be None" - - -def test_update_final_normalized_score(db): - """Test updating final normalized scores for UIDs.""" - db.insert_or_reset_uid(1, "hotkey1", 0.1, 0.0) - db.update_final_normalized_score(1, 0.99) - norm_score = db.get_normalized_score(1) - assert norm_score == 0.99, "Normalized score should be updated to 0.99" - - # Test updating non-existent UID (should not raise error, no change) - db.update_final_normalized_score(2, 0.88) - norm_score_2 = db.get_normalized_score(2) - assert norm_score_2 == 0.0, "Normalized score for non-existent UID 2 should be default 0.0" - - -def test_get_raw_eval_score(db): - """Test retrieving raw evaluation scores.""" - # Test non-existing UID - score = db.get_raw_eval_score(1) - assert score is None, "Non-existing UID should return None for raw score" - - # Insert UID and check raw score - db.insert_or_reset_uid(1, "hotkey1", 0.123, 0.0) - score = db.get_raw_eval_score(1) - assert score == 0.123, "Raw score should be 0.123" - - # Update raw score and check - db.update_raw_eval_score(1, 0.456) - score = db.get_raw_eval_score(1) - assert score == 0.456, "Raw score should be updated to 0.456" - -def test_get_all_normalized_scores(db): - """Test retrieving all normalized scores for a list of UIDs.""" - db.insert_or_reset_uid(1, "hotkey1", 0.1, 0.5) - db.insert_or_reset_uid(2, "hotkey2", 0.2, 0.6) - db.update_final_normalized_score(1, 0.55) - db.update_final_normalized_score(2, 0.66) - - # Get scores for existing UIDs - scores = db.get_all_normalized_scores([1, 2]) - assert len(scores) == 2, "Should return two scores" - assert scores[0] == 0.55, "UID 1 should have normalized score 0.55" - assert scores[1] == 0.66, "UID 2 should have normalized score 0.66" - - # Get score for non-existing UID - scores = db.get_all_normalized_scores([3]) - assert scores == [0.0], "Non-existing UID should return [0.0]" - - # Get scores for mixed UIDs - scores = db.get_all_normalized_scores([1, 3]) - assert len(scores) == 2, "Should return two scores" - assert scores[0] == 0.55, "Existing UID 1 should have normalized score 0.55" - assert scores[1] == 0.0, "Non-existing UID 3 should have normalized score 0.0" - - # Test with empty list - scores = db.get_all_normalized_scores([]) - assert scores == [], "Empty UID list should return empty list" - - -def test_get_normalized_score(db): - """Test retrieving a single normalized score for a UID.""" - # Test non-existing UID - score = db.get_normalized_score(1) - assert score == 0.0, "Non-existing UID should return 0.0 for normalized score" - - # Insert UID and check initial normalized score - db.insert_or_reset_uid(1, "hotkey1", 0.1, 0.05) - score = db.get_normalized_score(1) - assert score == 0.05, "Initial normalized score should be 0.05" - - # Update normalized score and check - db.update_final_normalized_score(1, 0.75) - score = db.get_normalized_score(1) - assert score == 0.75, "Normalized score should be updated to 0.75" - - # Test with multiple UIDs - db.insert_or_reset_uid(2, "hotkey2", 0.2, 0.15) - db.update_final_normalized_score(2, 0.85) - score1 = db.get_normalized_score(1) - score2 = db.get_normalized_score(2) - assert score1 == 0.75, "UID 1 normalized score should be 0.75" - assert score2 == 0.85, "UID 2 normalized score should be 0.85" + + submissions_updated = db.get_competition_submissions(comp_id) + assert submissions_updated[uid]['revision'] == "dev_branch", "Submission should be updated" + + +def test_record_submission_loss(db): + """Test updating the evaluation loss and eligibility.""" + comp_id = "comp_1" + uid = 10 + db.create_competition(comp_id, 1000, "commit_x") + + # Must exist before recording loss + db.record_submission(comp_id, uid, "hk", "ck", 1, 1, "ns", "rev") + + # 1. Record Loss + loss_val = 0.456 + is_eligible = True + db.record_submission_loss(comp_id, uid, loss_val, is_eligible) + + # 2. Verify + submissions = db.get_competition_submissions(comp_id) + assert submissions[uid]['eval_loss'] == loss_val + assert submissions[uid]['is_eligible'] == 1 # SQLite stores bool as 0/1 + + +def test_competition_winner_and_score(db): + """Test setting and retrieving the competition winner.""" + comp_id = "comp_winner_test" + db.create_competition(comp_id, 1000, "commit_x") + + winner_uid = 42 + winner_loss = 0.123 + + # 1. Update Score/Winner + db.update_competition_score(comp_id, winner_uid, winner_loss) + + # 2. Get Winner + retrieved_winner = db.get_competition_winner(comp_id) + assert retrieved_winner == winner_uid + + # Check directly via SQL that loss was saved + c = db.conn.cursor() + c.execute("SELECT winner_loss FROM daily_competitions WHERE competition_id=?", (comp_id,)) + assert c.fetchone()[0] == winner_loss + + +def test_dataset_revisions(db): + """Test setting and getting local dataset revisions.""" + namespace = "myset" + revision = "v1.0" + local_path = "/tmp/data" + + # 1. Set Revision + db.set_revision(namespace, revision, local_path) + + # 2. Get Revision + retrieved = db.get_revision(namespace, local_path) + assert retrieved == revision + + # 3. Update Revision + db.set_revision(namespace, "v1.1", local_path) + retrieved_updated = db.get_revision(namespace, local_path) + assert retrieved_updated == "v1.1" + + # 4. Non-existent + assert db.get_revision("other", "path") is None + + +def test_copy_competition_id(db): + """Test the functionality of copying/archiving a competition ID.""" + old_id = "comp_yesterday" + new_id = "comp_rewarding_record" + + # Setup initial competition + db.create_competition(old_id, 1000, "commit_old") + db.update_competition_score(old_id, winner_uid=7, winner_loss=0.5) + + # Perform Copy + db.copy_competition_id(new_id, old_id) + + # Verify New Entry + c = db.conn.cursor() + c.execute("SELECT dataset_commit, winner_uid, use_yesterday_reward, status FROM daily_competitions WHERE competition_id=?", (new_id,)) + row = c.fetchone() + + assert row is not None + assert row[0] == "commit_old" # Should copy commit + assert row[1] == 7 # Should copy winner + assert row[2] == 1 # use_yesterday_reward should be 1 + assert row[3] == "rewarding" # status should be 'rewarding' + + diff --git a/tests/FlockDataset/validators/test_scoring.py b/tests/FlockDataset/validators/test_scoring.py index 2518775..3d29aff 100644 --- a/tests/FlockDataset/validators/test_scoring.py +++ b/tests/FlockDataset/validators/test_scoring.py @@ -1,7 +1,10 @@ +import time +import pytest import numpy as np -from flockoff.validator.validator_utils import compute_score +from flockoff.validator.validator_utils import compute_score, select_winner from flockoff import constants - +from flockoff.validator.database import ScoreDB +from datetime import datetime, timezone DEFAULT_MIN_BENCH = 0.14 DEFAULT_MAX_BENCH = 0.2 @@ -100,24 +103,6 @@ def test_zero_benchmark_evaluation(): assert np.isclose(score, expected_score, rtol=1e-5) -def test_negative_benchmark_evaluation(): - loss = 0.1 - benchmark_loss = -0.1 - power = 2 - expected_score = constants.DEFAULT_NORMALIZED_SCORE - score = compute_score( - loss, - benchmark_loss, - DEFAULT_MIN_BENCH, - DEFAULT_MAX_BENCH, - power, - DEFAULT_BENCH_HEIGHT, - DEFAULT_COMPETITION_ID, - DEFAULT_COMPETITION_ID, - ) - assert np.isclose(score, expected_score, rtol=1e-5) - - def test_none_benchmark_evaluation(): loss = 0.1 benchmark_loss = None @@ -188,3 +173,55 @@ def test_mismatched_competition_id(): DEFAULT_COMPETITION_ID, ) assert np.isclose(score, expected_score, rtol=1e-5) + +@pytest.fixture +def db(): + """Fixture to create an in-memory database for each test.""" + db_instance = ScoreDB(":memory:") + yield db_instance + + +def test_select_winner(db): + score_db = db + now = datetime.now(timezone.utc) + competition_id_today = now.strftime("%Y%m%d") + score_db.record_submission(competition_id_today, 0, "h0", "c0", 100, int(time.time()), "name0", "revis0") + score_db.record_submission_loss(competition_id_today, 0, 0.101, is_eligible=True) + + score_db.record_submission(competition_id_today, 1, "h1", "c1", 101, int(time.time()), "name1", "revis1") + score_db.record_submission_loss(competition_id_today, 1, 0.1, is_eligible=True) + + score_db.record_submission(competition_id_today, 2, "h2", "c2", 102, int(time.time()), "name2", "revis2") + score_db.record_submission_loss(competition_id_today, 2, 10, is_eligible=True) + + score_db.record_submission(competition_id_today, 3, "h3", "c3", 103, int(time.time()), "name3", "revis3") + score_db.record_submission_loss(competition_id_today, 3, 3, is_eligible=True) + + temp_hot = {} + temp_cold = {} + for num in range(10, 255): + score_db.record_submission(competition_id_today, num, f"h{num}", f"c{num}", 100 + num, int(time.time()), f"name{num}", f"revis{num}") + score_db.record_submission_loss(competition_id_today, num, num, is_eligible=True) + temp_hot[num] = f"h{num}" + temp_cold[num] = f"c{num}" + + assert select_winner(score_db, competition_id_today, temp_hot | {0: "h0", 1: "h1", 2: "h2", 3: "h33"}, + temp_cold | {0: "c0", 1: "c1", 2: "c2", 3: "c33"}) == (0, 0.101) + + score_db.record_submission(competition_id_today, 4, "h4", "c4", 80, int(time.time()), "name4", "revis4") + score_db.record_submission_loss(competition_id_today, 4, 0.09, is_eligible=True) + + assert select_winner(score_db, competition_id_today, temp_hot | {0: "h0", 1: "h1", 2: "h2", 3: "h33", 4: "h44"}, + temp_cold | {0: "c0", 1: "c1", 2: "c2", 3: "c33", 4: "c44"}) == (0, 0.09) + + score_db.record_submission(competition_id_today, 5, "h5", "c4", 4, int(time.time()), "name5", "revis5") + score_db.record_submission_loss(competition_id_today, 5, 55, is_eligible=True) + + assert select_winner(score_db, competition_id_today, temp_hot | {0: "h0", 1: "h1", 2: "h2", 3: "h33", 4: "h44", 5: "h5"}, + temp_cold | {0: "c0", 1: "c1", 2: "c2", 3: "c33", 4: "c44", 5: "c4"}) == (5, 0.09) + + score_db.record_submission(competition_id_today, 5, "h5", "c4", 4, int(time.time()), "name5", "revis5") + score_db.record_submission_loss(competition_id_today, 5, 55, is_eligible=True) + + assert select_winner(score_db, competition_id_today, temp_hot | {0: "h0", 1: "h1", 2: "h2", 3: "h33", 4: "h44", 5: "h55"}, + temp_cold | {0: "c0", 1: "c1", 2: "c2", 3: "c33", 4: "c44", 5: "c45"}) == (0, 0.09) \ No newline at end of file