Rao sdk

bittensor/core/chain_data/__init__.py

@@ -17,6 +17,7 @@
 from .stake_info import StakeInfo
 from .subnet_hyperparameters import SubnetHyperparameters
 from .subnet_info import SubnetInfo
+from .subnet_state import SubnetState
 from .utils import custom_rpc_type_registry
 
 ProposalCallData = GenericCall

bittensor/core/chain_data/subnet_state.py

@@ -0,0 +1,90 @@
+"""
+This module defines the `SubnetState` data class and associated methods for handling and decoding
+subnetwork states in the Bittensor network.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+from scalecodec.utils.ss58 import ss58_encode
+
+from bittensor.core.chain_data.utils import (
+    ChainDataType,
+    from_scale_encoding,
+    SS58_FORMAT,
+)
+from bittensor.utils import u16_normalized_float
+from bittensor.utils.balance import Balance
+
+
+@dataclass
+class SubnetState:
+    netuid: int
+    hotkeys: list[str]
+    coldkeys: list[str]
+    active: list[bool]
+    validator_permit: list[bool]
+    pruning_score: list[float]
+    last_update: list[int]
+    emission: list[Balance]
+    dividends: list[float]
+    incentives: list[float]
+    consensus: list[float]
+    trust: list[float]
+    rank: list[float]
+    block_at_registration: list[int]
+    local_stake: list[Balance]
+    global_stake: list[Balance]
+    stake_weight: list[float]
+    emission_history: list[list[int]]
+
+    @classmethod
+    def from_vec_u8(cls, vec_u8: list[int]) -> Optional["SubnetState"]:
+        if len(vec_u8) == 0:
+            return None
+        decoded = from_scale_encoding(vec_u8, ChainDataType.SubnetState, is_option=True)
+        if decoded is None:
+            return None
+        return SubnetState.fix_decoded_values(decoded)
+
+    @classmethod
+    def list_from_vec_u8(cls, vec_u8: list[int]) -> list["SubnetState"]:
+        decoded = from_scale_encoding(
+            vec_u8, ChainDataType.SubnetState, is_vec=True, is_option=True
+        )
+        if decoded is None:
+            return []
+        decoded = [SubnetState.fix_decoded_values(d) for d in decoded]
+        return decoded
+
+    @classmethod
+    def fix_decoded_values(cls, decoded: dict) -> "SubnetState":
+        netuid = decoded["netuid"]
+        return SubnetState(
+            netuid=netuid,
+            hotkeys=[ss58_encode(val, SS58_FORMAT) for val in decoded["hotkeys"]],
+            coldkeys=[ss58_encode(val, SS58_FORMAT) for val in decoded["coldkeys"]],
+            active=decoded["active"],
+            validator_permit=decoded["validator_permit"],
+            pruning_score=[
+                u16_normalized_float(val) for val in decoded["pruning_score"]
+            ],
+            last_update=decoded["last_update"],
+            emission=[
+                Balance.from_rao(val).set_unit(netuid) for val in decoded["emission"]
+            ],
+            dividends=[u16_normalized_float(val) for val in decoded["dividends"]],
+            incentives=[u16_normalized_float(val) for val in decoded["incentives"]],
+            consensus=[u16_normalized_float(val) for val in decoded["consensus"]],
+            trust=[u16_normalized_float(val) for val in decoded["trust"]],
+            rank=[u16_normalized_float(val) for val in decoded["rank"]],
+            block_at_registration=decoded["block_at_registration"],
+            local_stake=[
+                Balance.from_rao(val).set_unit(netuid) for val in decoded["local_stake"]
+            ],
+            global_stake=[
+                Balance.from_rao(val).set_unit(0) for val in decoded["global_stake"]
+            ],
+            stake_weight=[u16_normalized_float(val) for val in decoded["stake_weight"]],
+            emission_history=decoded["emission_history"],
+        )

bittensor/core/chain_data/utils.py

@@ -22,6 +22,7 @@ class ChainDataType(Enum):
     SubnetHyperparameters = 8
     ScheduledColdkeySwapInfo = 9
     AccountId = 10
+    SubnetState = 11
 
 
 def from_scale_encoding(
@@ -208,6 +209,29 @@ def from_scale_encoding_using_type_string(
                 ["ip_type_and_protocol", "Compact<u8>"],
             ],
         },
+        "SubnetState": {
+            "type": "struct",
+            "type_mapping": [
+                ["netuid", "Compact<u16>"],
+                ["hotkeys", "Vec<AccountId>"],
+                ["coldkeys", "Vec<AccountId>"],
+                ["active", "Vec<bool>"],
+                ["validator_permit", "Vec<bool>"],
+                ["pruning_score", "Vec<Compact<u16>>"],
+                ["last_update", "Vec<Compact<u64>>"],
+                ["emission", "Vec<Compact<u64>>"],
+                ["dividends", "Vec<Compact<u16>>"],
+                ["incentives", "Vec<Compact<u16>>"],
+                ["consensus", "Vec<Compact<u16>>"],
+                ["trust", "Vec<Compact<u16>>"],
+                ["rank", "Vec<Compact<u16>>"],
+                ["block_at_registration", "Vec<Compact<u64>>"],
+                ["local_stake", "Vec<Compact<u64>>"],
+                ["global_stake", "Vec<Compact<u64>>"],
+                ["stake_weight", "Vec<Compact<u16>>"],
+                ["emission_history", "Vec<Vec<Compact<u64>>>"],
+            ],
+        },
         "StakeInfo": {
             "type": "struct",
             "type_mapping": [

bittensor/core/metagraph.py

@@ -25,16 +25,18 @@
 
 import numpy as np
 from numpy.typing import NDArray
+from substrateinterface.exceptions import SubstrateRequestException
 
+from bittensor.core import settings
+from bittensor.core.chain_data import AxonInfo, SubnetState
+from bittensor.utils.balance import Balance
 from bittensor.utils.btlogging import logging
 from bittensor.utils.registration import torch, use_torch
 from bittensor.utils.weight_utils import (
     convert_weight_uids_and_vals_to_tensor,
     convert_bond_uids_and_vals_to_tensor,
     convert_root_weight_uids_and_vals_to_tensor,
 )
-from . import settings
-from .chain_data import AxonInfo
 
 # For annotation purposes
 if typing.TYPE_CHECKING:
@@ -210,20 +212,66 @@ class MetagraphMixin(ABC):
     weights: Union["torch.nn.Parameter", NDArray]
     bonds: Union["torch.nn.Parameter", NDArray]
     uids: Union["torch.nn.Parameter", NDArray]
+    local_stake: Union["torch.nn.Parameter", NDArray]
+    global_stake: Union["torch.nn.Parameter", NDArray]
+    stake_weights: Union["torch.nn.Parameter", NDArray]
+    global_weight: Union["torch.nn.Parameter", NDArray]
     axons: list[AxonInfo]
 
     @property
-    def S(self) -> Union[NDArray, "torch.nn.Parameter"]:
+    def GS(self) -> list[Balance]:
         """
-        Represents the stake of each neuron in the Bittensor network. Stake is an important concept in the
-        Bittensor ecosystem, signifying the amount of network weight (or “stake”) each neuron holds,
-        represented on a digital ledger. The stake influences a neuron's ability to contribute to and benefit
-        from the network, playing a crucial role in the distribution of incentives and decision-making processes.
+        Represents the global stake of each neuron in the Bittensor network.
 
         Returns:
-            NDArray: A tensor representing the stake of each neuron in the network. Higher values signify a greater stake held by the respective neuron.
+            List[Balance]: The list of global stake of each neuron in the network.
         """
-        return self.total_stake
+        return self.global_stake
+
+    @property
+    def LS(self) -> list[Balance]:
+        """
+        Represents the local stake of each neuron in the Bittensor network.
+
+        Returns:
+            List[Balance]: The list of local stake of each neuron in the network.
+        """
+        return self.local_stake
+
+    @property
+    def SW(self) -> list[float]:
+        """
+        Represents the stake weights of each neuron in the Bittensor network.
+
+        Returns:
+            List[float]: Each value is calculated based on local_stake and global_stake.
+        """
+        return self.stake_weights
+
+    @property
+    def S(self) -> float:
+        """
+        Represents the value between 0.0 and 1.0. This gives the users do blacklists in terms of stake values.
+
+        Returns:
+            float: The value between 0.0 and 1.0 or None if stake_weights doesn't have zero index value.
+        """
+        try:
+            value = self.stake_weights[0] * max(self.global_stake).tao
+        except IndexError:
+            logging.warning("Stake weights is empty.")
+            value = None
+        return value
+
+    @property
+    def GW(self) -> float:
+        """
+        Represents Global Weights of subnet across all subnets.
+
+        Returns:
+            float: The value of Global Weights.
+        """
+        return self.global_weight
 
     @property
     def R(self) -> Union[NDArray, "torch.nn.Parameter"]:
@@ -578,6 +626,12 @@ def sync(
         if not lite:
             self._set_weights_and_bonds(subtensor=subtensor)
 
+        # Get global weight for netuid
+        self.global_weight = subtensor.get_global_weight(netuid=self.netuid)
+
+        # Fills in the stake associated attributes of a class instance from a chain response.
+        self._get_all_stakes_from_chain(subtensor=subtensor)
+
     def _initialize_subtensor(self, subtensor: "Subtensor"):
         """
         Initializes the subtensor to be used for syncing the metagraph.
@@ -748,6 +802,35 @@ def _process_weights_or_bonds(
     def _set_metagraph_attributes(self, block, subtensor):
         pass
 
+    def _get_all_stakes_from_chain(self, subtensor: "Subtensor"):
+        """Fills in the stake associated attributes of a class instance from a chain response."""
+        try:
+            hex_bytes_result = subtensor.query_runtime_api(
+                runtime_api="SubnetInfoRuntimeApi",
+                method="get_subnet_state",
+                params=[self.netuid],
+            )
+
+            if hex_bytes_result is None:
+                logging.debug(
+                    f"Unable to retrieve subnet state for netuid `{self.netuid}`."
+                )
+                return []
+
+            if hex_bytes_result.startswith("0x"):
+                bytes_result = bytes.fromhex(hex_bytes_result[2:])
+            else:
+                bytes_result = bytes.fromhex(hex_bytes_result)
+
+            subnet_state: "SubnetState" = SubnetState.from_vec_u8(bytes_result)
+            self.global_stake = subnet_state.global_stake
+            self.local_stake = subnet_state.local_stake
+            self.stake_weights = subnet_state.stake_weight
+        except (SubstrateRequestException, AttributeError):
+            logging.debug(
+                "Fields `global_stake`, `local_stake`, `stake_weights` can be obtained only from the RAO network."
+            )
+
     def _process_root_weights(
         self, data: list, attribute: str, subtensor: "Subtensor"
     ) -> Union[NDArray, "torch.nn.Parameter"]:
@@ -984,6 +1067,10 @@ def __init__(
         self.uids = torch.nn.Parameter(
             torch.tensor([], dtype=torch.int64), requires_grad=False
         )
+        self.local_stake: list[Balance] = []
+        self.global_stake: list[Balance] = []
+        self.stake_weights: list[float] = []
+        self.global_weight: Optional[float] = None
         self.axons: list[AxonInfo] = []
         if sync:
             self.sync(block=None, lite=lite)
@@ -1163,6 +1250,10 @@ def __init__(
         self.weights = np.array([], dtype=np.float32)
         self.bonds = np.array([], dtype=np.int64)
         self.uids = np.array([], dtype=np.int64)
+        self.local_stake: list[Balance] = []
+        self.global_stake: list[Balance] = []
+        self.stake_weights: list[float] = []
+        self.global_weight: Optional[float] = None
         self.axons: list[AxonInfo] = []
         if sync:
             self.sync(block=None, lite=lite)