Merge pull request #2683 from opentensor/feat/update-delegate-info

Updates DelegateInfo chain data

Author: ibraheem-abe

bittensor/core/chain_data/__init__.py

@@ -7,7 +7,7 @@
 
 from .axon_info import AxonInfo
 from .chain_identity import ChainIdentity
-from .delegate_info import DelegateInfo
+from .delegate_info import DelegateInfo, DelegatedInfo
 from .delegate_info_lite import DelegateInfoLite
 from .dynamic_info import DynamicInfo
 from .ip_info import IPInfo
@@ -36,6 +36,7 @@
     AxonInfo,
     ChainIdentity,
     DelegateInfo,
+    DelegatedInfo,
     DelegateInfoLite,
     DynamicInfo,
     IPInfo,

bittensor/core/chain_data/delegate_info.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Any, Optional
+from typing import Optional
 
 from bittensor.core.chain_data.info_base import InfoBase
 from bittensor.core.chain_data.utils import decode_account_id
@@ -8,27 +8,20 @@
 
 
 @dataclass
-class DelegateInfo(InfoBase):
-    """
-    Dataclass for delegate information. For a lighter version of this class, see ``DelegateInfoLite``.
+class DelegateInfoBase(InfoBase):
+    """Base class containing common delegate information fields.
 
-    Args:
-        hotkey_ss58 (str): Hotkey of the delegate for which the information is being fetched.
-        total_stake (int): Total stake of the delegate.
-        nominators (list[tuple[str, int]]): List of nominators of the delegate and their stake.
+    Attributes:
+        hotkey_ss58 (str): Hotkey of delegate.
+        owner_ss58 (str): Coldkey of owner.
         take (float): Take of the delegate as a percentage.
-        owner_ss58 (str): Coldkey of the owner.
-        registrations (list[int]): List of subnets that the delegate is registered on.
         validator_permits (list[int]): List of subnets that the delegate is allowed to validate on.
-        return_per_1000 (int): Return per 1000 TAO, for the delegate over a day.
-        total_daily_return (int): Total daily return of the delegate.
+        registrations (list[int]): List of subnets that the delegate is registered on.
+        return_per_1000 (Balance): Return per 1000 tao of the delegate over a day.
+        total_daily_return (Balance): Total daily return of the delegate.
     """
 
     hotkey_ss58: str  # Hotkey of delegate
-    total_stake: Balance  # Total stake of the delegate
-    nominators: list[
-        tuple[str, Balance]
-    ]  # List of nominators of the delegate and their stake
     owner_ss58: str  # Coldkey of owner
     take: float  # Take of the delegate as a percentage
     validator_permits: list[
@@ -38,35 +31,87 @@ class DelegateInfo(InfoBase):
     return_per_1000: Balance  # Return per 1000 tao of the delegate over a day
     total_daily_return: Balance  # Total daily return of the delegate
 
+
+@dataclass
+class DelegateInfo(DelegateInfoBase):
+    """
+    Dataclass for delegate information.
+
+    Additional Attributes:
+        total_stake (dict[int, Balance]): Total stake of the delegate mapped by netuid.
+        nominators (dict[str, dict[int, Balance]]): Mapping of nominator SS58 addresses to their stakes per subnet.
+    """
+
+    total_stake: dict[int, Balance]  # Total stake of the delegate by netuid and stake
+    nominators: dict[
+        str, dict[int, Balance]
+    ]  # Mapping of nominator addresses to their stakes per subnet
+
     @classmethod
     def _from_dict(cls, decoded: dict) -> Optional["DelegateInfo"]:
-        """Returns a DelegateInfo object from decoded chain data."""
-        nominators = [
-            (decode_account_id(x), Balance.from_rao(y))
-            for x, y in decoded["nominators"]
-        ]
-        total_stake = sum((x[1] for x in nominators)) if nominators else Balance(0)
-
-        return DelegateInfo(
-            hotkey_ss58=decode_account_id(decoded["delegate_ss58"]),
+        hotkey = decode_account_id(decoded.get("delegate_ss58"))
+        owner = decode_account_id(decoded.get("owner_ss58"))
+
+        nominators = {}
+        total_stake_by_netuid = {}
+
+        for raw_nominator, raw_stakes in decoded.get("nominators", []):
+            nominator_ss58 = decode_account_id(raw_nominator)
+            stakes = {
+                int(netuid): Balance.from_rao(stake_amt).set_unit(int(netuid))
+                for (netuid, stake_amt) in raw_stakes
+            }
+            nominators[nominator_ss58] = stakes
+
+            for netuid, stake in stakes.items():
+                if netuid not in total_stake_by_netuid:
+                    total_stake_by_netuid[netuid] = Balance(0).set_unit(netuid)
+                total_stake_by_netuid[netuid] += stake
+
+        return cls(
+            hotkey_ss58=hotkey,
+            total_stake=total_stake_by_netuid,
             nominators=nominators,
-            owner_ss58=decode_account_id(decoded["owner_ss58"]),
-            registrations=decoded["registrations"],
-            return_per_1000=Balance.from_rao(decoded["return_per_1000"]),
-            take=u16_normalized_float(decoded["take"]),
-            total_daily_return=Balance.from_rao(decoded["total_daily_return"]),
-            total_stake=total_stake,
-            validator_permits=decoded["validator_permits"],
+            owner_ss58=owner,
+            take=u16_normalized_float(decoded.get("take")),
+            validator_permits=list(decoded.get("validator_permits", [])),
+            registrations=list(decoded.get("registrations", [])),
+            return_per_1000=Balance.from_rao(decoded.get("return_per_1000")),
+            total_daily_return=Balance.from_rao(decoded.get("total_daily_return")),
         )
 
+
+@dataclass
+class DelegatedInfo(DelegateInfoBase):
+    """
+    Dataclass for delegated information. This class represents a delegate's information
+    specific to a particular subnet.
+
+    Additional Attributes:
+        netuid (int): Network ID of the subnet.
+        stake (Balance): Stake amount for this specific delegation.
+    """
+
+    netuid: int
+    stake: Balance
+
     @classmethod
-    def delegated_list_from_dicts(
-        cls, delegates: list[Any]
-    ) -> list[tuple["DelegateInfo", Balance]]:
-        return [
-            (
-                DelegateInfo.from_dict(delegate),
-                Balance.from_rao(balance),
-            )
-            for delegate, balance in delegates
-        ]
+    def _from_dict(
+        cls, decoded: tuple[dict, tuple[int, int]]
+    ) -> Optional["DelegatedInfo"]:
+        delegate_info, (netuid, stake) = decoded
+        hotkey = decode_account_id(delegate_info.get("delegate_ss58"))
+        owner = decode_account_id(delegate_info.get("owner_ss58"))
+        return cls(
+            hotkey_ss58=hotkey,
+            owner_ss58=owner,
+            take=u16_normalized_float(delegate_info.get("take")),
+            validator_permits=list(delegate_info.get("validator_permits", [])),
+            registrations=list(delegate_info.get("registrations", [])),
+            return_per_1000=Balance.from_rao(delegate_info.get("return_per_1000")),
+            total_daily_return=Balance.from_rao(
+                delegate_info.get("total_daily_return")
+            ),
+            netuid=int(netuid),
+            stake=Balance.from_rao(int(stake)).set_unit(int(netuid)),
+        )

bittensor/core/subtensor.py

@@ -24,6 +24,7 @@
     WeightCommitInfo,
     SubnetIdentity,
     SubnetInfo,
+    DelegatedInfo,
     decode_account_id,
 )
 from bittensor.core.chain_data.utils import decode_metadata
@@ -925,7 +926,7 @@ def get_delegated(
         if not result:
             return []
 
-        return DelegateInfo.delegated_list_from_dicts(result)
+        return DelegatedInfo.list_from_dicts(result)
 
     def get_delegates(self, block: Optional[int] = None) -> list["DelegateInfo"]:
         """