diff --git a/.env.miner.example b/.env.miner.example index 919c020d..1cf8b5f6 100644 --- a/.env.miner.example +++ b/.env.miner.example @@ -17,6 +17,7 @@ SUBTENSOR_ENDPOINT=wss://entrypoint-finney.opentensor.ai:443 # NETUID=98 # SUBTENSOR_NETWORK=test # SUBTENSOR_ENDPOINT=ws://testnet-lite:9944 +# VALIDATOR_MIN_STAKE=20000 # Task related config # this a maximum of 4 workers may submit responses for a single task diff --git a/.gitignore b/.gitignore index 8e193826..0205291e 100644 --- a/.gitignore +++ b/.gitignore @@ -188,3 +188,6 @@ testing/ # prisma database/prisma/ + +# scores data +scores/*.pt diff --git a/README.md b/README.md index b6d6c047..99871d29 100644 --- a/README.md +++ b/README.md @@ -278,6 +278,7 @@ DOJO_API_KEY= # blank for now WALLET_COLDKEY=# the name of the coldkey WALLET_HOTKEY=# the name of the hotkey AXON_PORT=8888 # port to serve requests over the public network for validators to call +VALIDATOR_MIN_STAKE=20000 # minimum stake required for validators default is 20000 TAO (use this to bypass the blacklist function in testnet) # Task related config TASK_MAX_RESULT=4 # this means that each miner can have up to 4 workers fill in responses ``` diff --git a/commons/objects.py b/commons/objects.py index 1d8ae1c9..bccce52f 100644 --- a/commons/objects.py +++ b/commons/objects.py @@ -11,10 +11,12 @@ class ObjectManager: def get_miner(cls): if get_config().simulation: from simulator.miner import MinerSim + if cls._miner is None: cls._miner = MinerSim() else: from neurons.miner import Miner + if cls._miner is None: cls._miner = Miner() return cls._miner @@ -23,10 +25,12 @@ def get_miner(cls): def get_validator(cls): if get_config().simulation: from simulator.validator import ValidatorSim + if cls._validator is None: cls._validator = ValidatorSim() else: from neurons.validator import Validator + if cls._validator is None: cls._validator = Validator() return cls._validator diff --git a/commons/score_storage.py b/commons/score_storage.py new file mode 100644 index 00000000..ec8d8c77 --- /dev/null +++ b/commons/score_storage.py @@ -0,0 +1,89 @@ +import json +from pathlib import Path + +import torch +from bittensor.btlogging import logging as logger + +from database.client import connect_db, disconnect_db +from database.prisma.models import Score_Model + + +class ScoreStorage: + """Handles persistence of validator scores""" + + SCORES_DIR = Path("scores") + SCORES_FILE = SCORES_DIR / "miner_scores.pt" + + @classmethod + async def migrate_from_db(cls) -> bool: + """One-time migration of scores from database to .pt file + Returns: + bool: True if migration successful or file already exists, False if migration failed + """ + try: + if cls.SCORES_FILE.exists(): + logger.info("Scores file already exists, skipping migration") + return True + + # Connect to database first + await connect_db() + + try: + # Get scores from database + score_record = await Score_Model.prisma().find_first( + order={"created_at": "desc"} + ) + if not score_record: + logger.warning("No scores found in database to migrate") + return True # Not an error, just no scores yet + + scores = torch.tensor(json.loads(score_record.score)) + + # Create scores directory if it doesn't exist + cls.SCORES_DIR.mkdir(exist_ok=True) + + # Save scores to .pt file + torch.save(scores, cls.SCORES_FILE) + logger.success(f"Successfully migrated scores to {cls.SCORES_FILE}") + + # Verify the migration + loaded_scores = torch.load(cls.SCORES_FILE) + if torch.equal(scores, loaded_scores): + logger.success("Migration verification successful - scores match") + return True + else: + logger.error("Migration verification failed - scores do not match") + return False + + finally: + await disconnect_db() + + except Exception as e: + logger.error(f"Failed to migrate scores: {e}") + return False + + @classmethod + async def save(cls, scores: torch.Tensor) -> None: + """Save validator scores to .pt file""" + try: + cls.SCORES_DIR.mkdir(exist_ok=True) + torch.save(scores, cls.SCORES_FILE) + logger.success("Successfully saved validator scores to file") + except Exception as e: + logger.error(f"Failed to save validator scores: {e}") + raise + + @classmethod + async def load(cls) -> torch.Tensor | None: + """Load validator scores from .pt file""" + try: + if not cls.SCORES_FILE.exists(): + logger.warning("No validator scores file found") + return None + + scores = torch.load(cls.SCORES_FILE) + logger.success("Successfully loaded validator scores from file") + return scores + except Exception as e: + logger.error(f"Failed to load validator scores: {e}") + return None diff --git a/dojo/__init__.py b/dojo/__init__.py index 5674629d..aa3edec3 100644 --- a/dojo/__init__.py +++ b/dojo/__init__.py @@ -30,7 +30,7 @@ def get_latest_git_tag(): ) -VALIDATOR_MIN_STAKE = 20000 +VALIDATOR_MIN_STAKE = int(os.getenv("VALIDATOR_MIN_STAKE", "20000")) TASK_DEADLINE = 6 * 60 * 60 # Define the time intervals for various tasks. @@ -44,7 +44,7 @@ def get_latest_git_tag(): if get_config().fast_mode: print("Running in fast mode for testing purposes...") - VALIDATOR_MIN_STAKE = 20000 + VALIDATOR_MIN_STAKE = int(os.getenv("VALIDATOR_MIN_STAKE", "20000")) TASK_DEADLINE = 180 VALIDATOR_RUN = 60 VALIDATOR_HEARTBEAT = 15 diff --git a/entrypoints.sh b/entrypoints.sh index 325fc7ad..33001a25 100755 --- a/entrypoints.sh +++ b/entrypoints.sh @@ -73,4 +73,4 @@ if [ "$1" = 'validator' ]; then --neuron.type validator \ --wandb.project_name ${WANDB_PROJECT_NAME} \ ${EXTRA_ARGS} -fi \ No newline at end of file +fi diff --git a/neurons/validator.py b/neurons/validator.py index 0d0e620c..34222911 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -34,6 +34,7 @@ from commons.obfuscation.obfuscation_utils import obfuscate_html_and_js from commons.objects import ObjectManager from commons.orm import ORM +from commons.score_storage import ScoreStorage from commons.scoring import Scoring from commons.utils import ( _terminal_plot, @@ -45,7 +46,6 @@ set_expire_time, ttl_get_block, ) -from database.client import connect_db from dojo import __spec_version__ from dojo.protocol import ( CompletionResponses, @@ -103,10 +103,17 @@ def __init__(self): self.scores: torch.Tensor = torch.zeros( len(self.metagraph.hotkeys), dtype=torch.float32 ) - # manually always register and always sync metagraph when application starts self.check_registered() - self.executor = ThreadPoolExecutor(max_workers=2) + # Run score migration before loading state + migration_success = self.loop.run_until_complete(ScoreStorage.migrate_from_db()) + if not migration_success: + logger.error( + "Score migration failed - cannot continue without valid scores" + ) + raise RuntimeError("Score migration failed - validator cannot start") + + self.executor = ThreadPoolExecutor(max_workers=2) self.load_state() init_wandb(config=self.config, my_uid=self.uid, wallet=self.wallet) @@ -713,7 +720,8 @@ async def save_state( logger.warning("Scores are all zeros, but saving anyway!") # raise EmptyScores("Skipping save as scores are all empty") - await ORM.create_or_update_validator_score(self.scores) + # await ORM.create_or_update_validator_score(self.scores) + await ScoreStorage.save(self.scores) logger.success(f"📦 Saved validator state with scores: {self.scores}") except EmptyScores as e: logger.debug(f"No need to to save validator state: {e}") @@ -722,8 +730,7 @@ async def save_state( async def _load_state(self): try: - await connect_db() - scores = await ORM.get_validator_score() + scores = await ScoreStorage.load() if scores is None: num_processed_tasks = await ORM.get_num_processed_tasks() diff --git a/scripts/inspect_scores.py b/scripts/inspect_scores.py new file mode 100644 index 00000000..fb04577b --- /dev/null +++ b/scripts/inspect_scores.py @@ -0,0 +1,78 @@ +import argparse +from typing import List + +import torch +from tabulate import tabulate +from termcolor import colored + + +def format_score_table(scores: torch.Tensor) -> List[List[str]]: + """Format scores into a table with 10 columns""" + table_data = [] + row = [] + for i, score in enumerate(scores): + score_str = f"{score.item():.4f}" + # Color non-zero scores + if score.item() > 0: + score_str = colored(score_str, "green") + row.append([i, score_str]) + + # Use 10 columns for better screen fit + if len(row) == 10 or i == len(scores) - 1: + table_data.append(row) + row = [] + return table_data + + +def inspect_scores( + file_path: str = "scores/validator_scores.pt", show_all: bool = False +): + try: + scores = torch.load(file_path) + + # Print Summary + print(colored("\n=== Scores Summary ===", "blue", attrs=["bold"])) + print(f"Total UIDs: {len(scores)}") + print(f"Data type: {scores.dtype}") + print(f"Device: {scores.device}") + + # Print Statistics + print(colored("\n=== Statistics ===", "blue", attrs=["bold"])) + print(f"Mean score: {scores.mean().item():.4f}") + print(f"Min score: {scores.min().item():.4f}") + print(f"Max score: {scores.max().item():.4f}") + print( + f"Non-zero UIDs: {torch.count_nonzero(scores).item()} " + f"({(torch.count_nonzero(scores).item()/len(scores)*100):.1f}%)" + ) + + # Print Top Scores + top_k = 10 # Show top 10 + values, indices = torch.topk(scores, k=min(top_k, len(scores))) + print(colored("\n=== Top 10 Scores ===", "blue", attrs=["bold"])) + top_scores = [ + [f"UID {idx}", f"{val.item():.4f}"] for idx, val in zip(indices, values) + ] + print(tabulate(top_scores, headers=["UID", "Score"], tablefmt="simple")) + + if show_all: + print(colored("\n=== All Scores ===", "blue", attrs=["bold"])) + table_data = format_score_table(scores) + for row in table_data: + # headers = [f"UID {i[0]}" for i in row] + values = [f"UID {i[0]} - {i[1]}" for i in row] + print(tabulate([values], tablefmt="simple")) + + print("\nNote: Green values indicate non-zero scores") + + except FileNotFoundError: + print(colored(f"Score file not found at {file_path}", "red")) + except Exception as e: + print(colored(f"Error reading scores: {e}", "red")) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--all", action="store_true", help="Show all scores") + args = parser.parse_args() + inspect_scores(show_all=args.all) diff --git a/simulator/miner.py b/simulator/miner.py index 2f58e5c8..c72dd5eb 100644 --- a/simulator/miner.py +++ b/simulator/miner.py @@ -22,10 +22,7 @@ def __init__(self): host = os.getenv("REDIS_HOST", "localhost") port = int(os.getenv("REDIS_PORT", 6379)) self.redis_client = redis.Redis( - host=host, - port=port, - db=0, - decode_responses=True + host=host, port=port, db=0, decode_responses=True ) logger.info("Redis connection established") @@ -42,12 +39,14 @@ def __init__(self): def _configure_simulation(self): """Configure simulation parameters with environment variables or defaults.""" self.response_behaviors = { - 'normal': float(os.getenv("SIM_NORMAL_RESP_PROB", 0.8)), - 'no_response': float(os.getenv("SIM_NO_RESP_PROB", 0.1)), - 'timeout': float(os.getenv("SIM_TIMEOUT_PROB", 0.1)) + "normal": float(os.getenv("SIM_NORMAL_RESP_PROB", 0.8)), + "no_response": float(os.getenv("SIM_NO_RESP_PROB", 0.1)), + "timeout": float(os.getenv("SIM_TIMEOUT_PROB", 0.1)), } - async def forward_feedback_request(self, synapse: FeedbackRequest) -> FeedbackRequest: + async def forward_feedback_request( + self, synapse: FeedbackRequest + ) -> FeedbackRequest: try: # Validate that synapse, dendrite, dendrite.hotkey, and response are not None if not synapse or not synapse.dendrite or not synapse.dendrite.hotkey: @@ -69,7 +68,7 @@ async def forward_feedback_request(self, synapse: FeedbackRequest) -> FeedbackRe self.redis_client.set( redis_key, new_synapse.model_dump_json(), - ex=86400 # expire after 24 hours + ex=86400, # expire after 24 hours ) logger.info(f"Stored feedback request {synapse.request_id}") @@ -81,7 +80,9 @@ async def forward_feedback_request(self, synapse: FeedbackRequest) -> FeedbackRe traceback.print_exc() return synapse - async def forward_task_result_request(self, synapse: TaskResultRequest) -> TaskResultRequest | None: + async def forward_task_result_request( + self, synapse: TaskResultRequest + ) -> TaskResultRequest | None: try: logger.info(f"Received TaskResultRequest for task id: {synapse.task_id}") if not synapse or not synapse.task_id: @@ -91,9 +92,9 @@ async def forward_task_result_request(self, synapse: TaskResultRequest) -> TaskR # Simulate different response behaviors behavior = self._get_response_behavior() - if behavior in ['no_response', 'timeout']: + if behavior in ["no_response", "timeout"]: logger.debug(f"Simulating {behavior} for task {synapse.task_id}") - if behavior == 'timeout': + if behavior == "timeout": await asyncio.sleep(30) return None @@ -113,17 +114,17 @@ async def forward_task_result_request(self, synapse: TaskResultRequest) -> TaskR for criteria_type in feedback_request.criteria_types: result = Result( type=criteria_type.type, - value=self._generate_scores(feedback_request.ground_truth) + value=self._generate_scores(feedback_request.ground_truth), ) task_result = TaskResult( id=get_new_uuid(), - status='COMPLETED', + status="COMPLETED", created_at=current_time, updated_at=current_time, result_data=[result], worker_id=get_new_uuid(), - task_id=synapse.task_id + task_id=synapse.task_id, ) task_results.append(task_result) @@ -144,7 +145,7 @@ def _get_response_behavior(self) -> str: """Determine the response behavior based on configured probabilities.""" return random.choices( list(self.response_behaviors.keys()), - weights=list(self.response_behaviors.values()) + weights=list(self.response_behaviors.values()), )[0] def _generate_scores(self, ground_truth: dict) -> dict: diff --git a/simulator/validator.py b/simulator/validator.py index 75ade66c..3982aa9c 100644 --- a/simulator/validator.py +++ b/simulator/validator.py @@ -48,8 +48,12 @@ def block(self): except BrokenPipeError: self._block_check_attempts += 1 if self._block_check_attempts >= self.MAX_BLOCK_CHECK_ATTEMPTS: - logger.error("Multiple failed attempts to get block number, attempting reconnection") - if asyncio.get_event_loop().run_until_complete(self._try_reconnect_subtensor()): + logger.error( + "Multiple failed attempts to get block number, attempting reconnection" + ) + if asyncio.get_event_loop().run_until_complete( + self._try_reconnect_subtensor() + ): return self.block return self._last_block if self._last_block is not None else 0 @@ -60,8 +64,8 @@ def block(self): def check_registered(self): new_subtensor = bt.subtensor(self.subtensor.config) if not new_subtensor.is_hotkey_registered( - netuid=self.config.netuid, - hotkey_ss58=self.wallet.hotkey.ss58_address, + netuid=self.config.netuid, + hotkey_ss58=self.wallet.hotkey.ss58_address, ): logger.error( f"Wallet: {self.wallet} is not registered on netuid {self.config.netuid}." @@ -70,9 +74,9 @@ def check_registered(self): exit() async def send_request( - self, - synapse: FeedbackRequest | None = None, - external_user: bool = False, + self, + synapse: FeedbackRequest | None = None, + external_user: bool = False, ): start = get_epoch_time() # typically the request may come from an external source however, @@ -89,7 +93,7 @@ async def send_request( self.metagraph.axons[uid] for uid in sel_miner_uids if self.metagraph.axons[uid].hotkey.casefold() - != self.wallet.hotkey.ss58_address.casefold() + != self.wallet.hotkey.ss58_address.casefold() ] if not len(axons): logger.warning("🤷 No axons to query ... skipping") @@ -135,7 +139,7 @@ async def send_request( prompt=data.prompt, completion_responses=data.responses, expire_at=expire_at, - ground_truth=data.ground_truth # Added ground truth!!!!! + ground_truth=data.ground_truth, # Added ground truth!!!!! ) elif external_user: obfuscated_model_to_model = self.obfuscate_model_names( @@ -217,24 +221,24 @@ async def send_request( @staticmethod async def _send_shuffled_requests( - dendrite: bt.dendrite, axons: List[bt.AxonInfo], synapse: FeedbackRequest + dendrite: bt.dendrite, axons: List[bt.AxonInfo], synapse: FeedbackRequest ) -> list[FeedbackRequest]: """Send the same request to all miners without shuffling the order. - WARNING: This should only be used for testing/debugging as it could allow miners to game the system. + WARNING: This should only be used for testing/debugging as it could allow miners to game the system. - Args: - dendrite (bt.dendrite): Communication channel to send requests - axons (List[bt.AxonInfo]): List of miner endpoints - synapse (FeedbackRequest): The feedback request to send + Args: + dendrite (bt.dendrite): Communication channel to send requests + axons (List[bt.AxonInfo]): List of miner endpoints + synapse (FeedbackRequest): The feedback request to send - Returns: - list[FeedbackRequest]: List of miner responses - """ + Returns: + list[FeedbackRequest]: List of miner responses + """ all_responses = [] batch_size = 10 for i in range(0, len(axons), batch_size): - batch_axons = axons[i: i + batch_size] + batch_axons = axons[i : i + batch_size] tasks = [] for axon in batch_axons: