ZANGIEF
""" HEADER = """This leaderboard showcases the top-performing miners in the Zangief - CommuneAI Translation Subnet. The models are ranked based on their daily rewards.
Zangief is a subnet dedicated to language translation. The goal of the subnet is to collectively bootstrap a language translation application that supports dozens of different languages, communication styles, and specific areas of expertise.
The actors that power the subnet are the miners and validators. The validators generate source material to be translated and pass the source material to the miners. The miners run web services that respond to the given source input with high quality translation. The miners also respond to queries that are served from an end-user application. Over time, the validators will also curate high quality translations to the source material which itself will be cleaned and compiled into a dataset. The dataset that is produced from the mining and validating activity on the subnet will be open source. This dataset can be used to train models or provide useful translations for subtitles or other online media.
""" EVALUATION_HEADER = """Name represents the model name. Rewards / Day indicates the expected daily rewards for each model in $COMAI. UID is the unique identifier of the miner. $USD Value is the estimated dollar value of the daily rewards.
""" LANGUAGES_SUPPORTED = """Arabic, Chinese, English, French, German, Hebrew, Hindi, Portuguese, Russian, Spanish, Urdu, Vietnamese, and more to come!
""" netuid = 23 node_url = "wss://commune-api-node-2.communeai.net" async def get_com_price(session: aiohttp.ClientSession) -> Any: retries = 5 for i in range(retries): try: async with session.get("https://api.mexc.com/api/v3/avgPrice?symbol=COMAIUSDT") as response: response.raise_for_status() price = float((await response.json())["price"]) print(f"Fetched COM price: {price}") return price except Exception as e: print(f"Error fetching COM price: {e}") await asyncio.sleep(retries) raise RuntimeError("Failed to fetch COM price") async def make_query(client: CommuneClient) -> tuple[dict[int, int], dict[int, str]]: request_dict = { "SubspaceModule": [ ("Name", [netuid]), ("Emission", []), ("Incentive", []), ("Dividends", []), ], } emission_dict = {} name_dict = {} result = client.query_batch_map(request_dict) emission = result["Emission"] netuid_emission = emission[netuid] incentive = result["Incentive"] netuid_incentive = incentive[netuid] dividends = result["Dividends"] netuid_dividends = dividends[netuid] names = result["Name"] highest_uid = max(names.keys()) for uid in range(highest_uid + 1): emission = netuid_emission[uid] if emission != 0: incentive = netuid_incentive[uid] dividends = netuid_dividends[uid] if incentive > 0: emission_dict[uid] = netuid_emission[uid] name_dict[uid] = names[uid] return emission_dict, name_dict async def get_leaderboard_data(): async with aiohttp.ClientSession() as session: try: com_price = await get_com_price(session) blocks_in_day = 10_800 client = CommuneClient(node_url) emission_dict, name_dict = await make_query(client) print("got the emission") scores = {} for uid, emi in emission_dict.items(): scores[uid] = (emi / 10**11) * blocks_in_day sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True) leaderboard_data = [] for rank, (uid, score) in enumerate(sorted_scores, start=1): name = name_dict[uid] units = score usd_value = score * com_price leaderboard_data.append((rank, uid, name, units, f"${usd_value:.2f}")) return leaderboard_data except Exception as e: print(f"Error fetching leaderboard data: {e}") return [] async def update_leaderboard_table(): leaderboard_data = await get_leaderboard_data() leaderboard_data = [list(row) for row in leaderboard_data] for row in leaderboard_data: row[0] = f"{row[0]} 🏆" total_usd_value = sum(float(row[4][1:]) for row in leaderboard_data) rewards_per_week = total_usd_value * 7 rewards_per_month = total_usd_value * 30 return leaderboard_data, f'''