Skip to content

Ethereum

Block

Bases: APIResource

Source code in src/rated/ethereum/blocks.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class Block(APIResource):
    path = "/blocks"

    def get(self, slot: int) -> EthBlock:
        """
        Get a block by consensus slot number

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> block = eth.block.get(500)
            >>> print(f"{block.total_rewards = }")

        Args:
            slot: Consensus slot number

        Returns:
            A single block
        """
        data = self.client.get(f"{self.resource_path}/{slot}")
        return json_to_instance(data, EthBlock)

get(slot)

Get a block by consensus slot number

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> block = eth.block.get(500)
>>> print(f"{block.total_rewards = }")

Parameters:

Name Type Description Default
slot int

Consensus slot number

 required

Returns:

Type Description
Block

A single block
Source code in src/rated/ethereum/blocks.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def get(self, slot: int) -> EthBlock:
    """
    Get a block by consensus slot number

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> block = eth.block.get(500)
        >>> print(f"{block.total_rewards = }")

    Args:
        slot: Consensus slot number

    Returns:
        A single block
    """
    data = self.client.get(f"{self.resource_path}/{slot}")
    return json_to_instance(data, EthBlock)

Blocks

Bases: APIResource

Blocks allows one to dig into the individual slots and blocks in Ethereum since the Merge. Metrics include validator rewards from the consensus and execution layers, MEV data, and missed reward estimates.

Source code in src/rated/ethereum/blocks.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class Blocks(APIResource):
    """
    Blocks allows one to dig into the individual slots and blocks in Ethereum since the Merge.
    Metrics include validator rewards from the consensus and execution layers, MEV data, and missed reward estimates.
    """

    path = "/blocks"

    def all(
        self,
        *,
        from_slot: int | None = None,
        size: int | None = None,
        follow_next: bool = False,
    ) -> Iterator[EthBlock]:
        """
        Get all blocks

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> blocks = eth.blocks.all(from_slot=500, size=10)
            >>> for block in blocks:
            >>>     print(f"{block.total_rewards = }")

        Args:
            from_slot: Start slot
            size: Number of results included per page
            follow_next: Whether to follow pagination or not

        Yields:
            An iterator over all blocks
        """
        params: Dict[str, Any] = {"from": from_slot, "size": size}
        return self.client.yield_paginated_results(
            self.resource_path,
            params=params,
            cls=EthBlock,
            follow_next=follow_next,
        )

all(*, from_slot=None, size=None, follow_next=False)

Get all blocks

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> blocks = eth.blocks.all(from_slot=500, size=10)
>>> for block in blocks:
>>>     print(f"{block.total_rewards = }")

Parameters:

Name Type Description Default
from_slot int | None

Start slot

 None
size int | None

Number of results included per page

 None
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
Block

An iterator over all blocks
Source code in src/rated/ethereum/blocks.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def all(
    self,
    *,
    from_slot: int | None = None,
    size: int | None = None,
    follow_next: bool = False,
) -> Iterator[EthBlock]:
    """
    Get all blocks

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> blocks = eth.blocks.all(from_slot=500, size=10)
        >>> for block in blocks:
        >>>     print(f"{block.total_rewards = }")

    Args:
        from_slot: Start slot
        size: Number of results included per page
        follow_next: Whether to follow pagination or not

    Yields:
        An iterator over all blocks
    """
    params: Dict[str, Any] = {"from": from_slot, "size": size}
    return self.client.yield_paginated_results(
        self.resource_path,
        params=params,
        cls=EthBlock,
        follow_next=follow_next,
    )

Network

Bases: APIResource

Network allows querying into a collection of statistics that provide an overview of the whole network in historical states, which can be as recent as the last 24h.

Source code in src/rated/ethereum/network.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class Network(APIResource):
    """
    Network allows querying into a collection of statistics that provide an overview of the whole network in historical
    states, which can be as recent as the last 24h.
    """

    path = "/network"

    def stats(self) -> Iterator[NetworkStats]:
        """
        Summarizes key performance statistics for all the whole network, for the current calendar day.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> network_stats = eth.network.stats()
            >>> for stat in network_stats:
            >>>     print(f"{stat.avg_uptime = }")

        Yields:
            Network performance stats
        """
        data = self.client.get(f"{self.resource_path}/stats")
        for item in data:
            yield json_to_instance(item, NetworkStats)

    def overview(self) -> Iterator[NetworkOverview]:
        """
        Summarizes key statistics for the whole network.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> overview = eth.network.overview()
            >>> for values in overview:
            >>>     print(f"{values.activating_validators = }")

        Yields:
            Statistics summary
        """
        data = self.client.get(f"{self.resource_path}/overview")
        for item in data:
            yield json_to_instance(item, NetworkOverview)

    def capacity(self) -> Iterator[NetworkChurnCapacity]:
        """
        Summarizes activations and exits for all the whole network.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> capacity = eth.network.capacity()
            >>> for cap in capacity:
            >>>     print(f"{cap.churn_limit = }")

        Yields:
            Activations and exits summary
        """
        data = self.client.get(f"{self.resource_path}/capacity")
        for item in data:
            yield json_to_instance(item, NetworkChurnCapacity)

    def capacity_pool(
        self,
        *,
        stake_action: StakeAction = StakeAction.ACTIVATION,
        time_window: TimeWindow = TimeWindow.ONE_DAY,
    ) -> Iterator[NetworkChurnCapacityPool]:
        """
        Summarizes activations and exits, broken down by staking pool.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> capacity_pool = eth.network.capacity_pool(time_window=TimeWindow.THIRTY_DAYS)
            >>> for cap in capacity_pool:
            >>>     print(f"{cap.churn_limit = }")

        Args:
            stake_action: Direction of flow. This can be either of activation or exit.
            time_window: The time window of aggregation. You might ask for 1d, 7d, 30d or All-time data

        Yields:
            An iterator over all results
        """
        params: Dict[str, Any] = {
            "stakeAction": stake_action.value,
            "window": time_window.value,
        }
        data = self.client.get(f"{self.resource_path}/capacity/pool", params=params)
        for item in data:
            yield json_to_instance(item, NetworkChurnCapacityPool)

capacity()

Summarizes activations and exits for all the whole network.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> capacity = eth.network.capacity()
>>> for cap in capacity:
>>>     print(f"{cap.churn_limit = }")

Yields:

Type Description
NetworkChurnCapacity

Activations and exits summary
Source code in src/rated/ethereum/network.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def capacity(self) -> Iterator[NetworkChurnCapacity]:
    """
    Summarizes activations and exits for all the whole network.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> capacity = eth.network.capacity()
        >>> for cap in capacity:
        >>>     print(f"{cap.churn_limit = }")

    Yields:
        Activations and exits summary
    """
    data = self.client.get(f"{self.resource_path}/capacity")
    for item in data:
        yield json_to_instance(item, NetworkChurnCapacity)

capacity_pool(*, stake_action=StakeAction.ACTIVATION, time_window=TimeWindow.ONE_DAY)

Summarizes activations and exits, broken down by staking pool.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> capacity_pool = eth.network.capacity_pool(time_window=TimeWindow.THIRTY_DAYS)
>>> for cap in capacity_pool:
>>>     print(f"{cap.churn_limit = }")

Parameters:

Name Type Description Default
stake_action StakeAction

Direction of flow. This can be either of activation or exit.

 ACTIVATION
time_window TimeWindow

The time window of aggregation. You might ask for 1d, 7d, 30d or All-time data

 ONE_DAY

Yields:

Type Description
NetworkChurnCapacityPool

An iterator over all results
Source code in src/rated/ethereum/network.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def capacity_pool(
    self,
    *,
    stake_action: StakeAction = StakeAction.ACTIVATION,
    time_window: TimeWindow = TimeWindow.ONE_DAY,
) -> Iterator[NetworkChurnCapacityPool]:
    """
    Summarizes activations and exits, broken down by staking pool.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> capacity_pool = eth.network.capacity_pool(time_window=TimeWindow.THIRTY_DAYS)
        >>> for cap in capacity_pool:
        >>>     print(f"{cap.churn_limit = }")

    Args:
        stake_action: Direction of flow. This can be either of activation or exit.
        time_window: The time window of aggregation. You might ask for 1d, 7d, 30d or All-time data

    Yields:
        An iterator over all results
    """
    params: Dict[str, Any] = {
        "stakeAction": stake_action.value,
        "window": time_window.value,
    }
    data = self.client.get(f"{self.resource_path}/capacity/pool", params=params)
    for item in data:
        yield json_to_instance(item, NetworkChurnCapacityPool)

overview()

Summarizes key statistics for the whole network.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> overview = eth.network.overview()
>>> for values in overview:
>>>     print(f"{values.activating_validators = }")

Yields:

Type Description
NetworkOverview

Statistics summary
Source code in src/rated/ethereum/network.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def overview(self) -> Iterator[NetworkOverview]:
    """
    Summarizes key statistics for the whole network.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> overview = eth.network.overview()
        >>> for values in overview:
        >>>     print(f"{values.activating_validators = }")

    Yields:
        Statistics summary
    """
    data = self.client.get(f"{self.resource_path}/overview")
    for item in data:
        yield json_to_instance(item, NetworkOverview)

stats()

Summarizes key performance statistics for all the whole network, for the current calendar day.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> network_stats = eth.network.stats()
>>> for stat in network_stats:
>>>     print(f"{stat.avg_uptime = }")

Yields:

Type Description
NetworkStats

Network performance stats
Source code in src/rated/ethereum/network.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def stats(self) -> Iterator[NetworkStats]:
    """
    Summarizes key performance statistics for all the whole network, for the current calendar day.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> network_stats = eth.network.stats()
        >>> for stat in network_stats:
        >>>     print(f"{stat.avg_uptime = }")

    Yields:
        Network performance stats
    """
    data = self.client.get(f"{self.resource_path}/stats")
    for item in data:
        yield json_to_instance(item, NetworkStats)

Operator

Bases: APIResource

Querying into pre-materialized operator groupings.

Source code in src/rated/ethereum/operators.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
class Operator(APIResource):
    """Querying into pre-materialized operator groupings."""

    path = "/operators"

    def effectiveness(
        self,
        operator_id: str,
        id_type: IdType,
        *,
        from_day: int | date | None = None,
        size: int | None = None,
        granularity: Granularity = Granularity.DAY,
        filter_type: FilterType = FilterType.DAY,
        follow_next: bool = False,
    ) -> Iterator[OperatorEffectiveness]:
        """
        Historical performance of a single operator.
        This includes rewards (aggregate and granular), performance (effectiveness and its components),
        slashing history and much more.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> effectiveness = eth.operator.effectiveness("Lido", id_type=IdType.POOL, from_day=795, size=10)
            >>> for eff in effectiveness:
            >>>     print(f"{eff.avg_validator_effectiveness = }, {eff.day = }")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class
            from_day: Start day
            size: Number of results included per page
            granularity:T he size of time increments you are looking to query
            filter_type: Hour, day and datetime
            follow_next: Whether to follow pagination or not

        Yields:
            Operator Effectiveness
        """
        url: str = f"{self.resource_path}/{operator_id}/effectiveness"
        params: Dict[str, Any] = {
            "idType": id_type.value,
            "from": from_day,
            "size": size,
            "granularity": granularity.value,
            "filterType": filter_type.value,
        }
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=OperatorEffectiveness,
            follow_next=follow_next,
        )

    def metadata(self, operator_id: str, id_type: IdType) -> OperatorType:
        """
        Retrieve profile information on specific operators.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> op = eth.operator.metadata("Lido", id_type=IdType.POOL)
            >>> print(f"{op.node_operator_count = }")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class

        Returns:
            Operator metadata.
        """
        url: str = f"{self.resource_path}/{operator_id}"
        params: Dict[str, Any] = {"idType": id_type.value}
        operator = self.client.get(url, params=params)
        return json_to_instance(operator, OperatorType)

    def clients(self, operator_id: str, id_type: IdType) -> Iterator[ClientPercentage]:
        """
        Consensus client distribution

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> clients = eth.operator.clients("Lido", id_type=IdType.POOL)
            >>> for c in clients:
            >>>     print(f"{c.client = }, {c.percentage = }%")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class

        Yields:
            Clients percentages
        """
        url: str = f"{self.resource_path}/{operator_id}/clients"
        params: Dict[str, Any] = {"idType": id_type.value}
        data = self.client.get(url, params=params)
        for item in data:
            yield json_to_instance(item, ClientPercentage)

    def relayers(
        self,
        operator_id: str,
        id_type: IdType,
        *,
        time_window: TimeWindow = TimeWindow.THIRTY_DAYS,
    ) -> Iterator[RelayerPercentage]:
        """
        Get information relating to an entity's historical distribution of relays they have procured blocks from.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> relayers = eth.operator.relayers("Lido", id_type=IdType.POOL, time_window=TimeWindow.ALL_TIME)
            >>> for r in relayers:
            >>>     print(f"{r.relayer = }, {r.percentage = }%")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class
            time_window: The time window of aggregation

        Yields:
            Relayer Percentages
        """
        url: str = f"{self.resource_path}/{operator_id}/relayers"
        params: Dict[str, Any] = {"idType": id_type.value, "window": time_window.value}
        data = self.client.get(url, params=params)
        for item in data:
            yield json_to_instance(item, RelayerPercentage)

    def apr(
        self,
        operator_id: str,
        id_type: IdType,
        *,
        time_window: TimeWindow,
        apr_type: AprType = AprType.BACKWARD,
    ) -> OperatorApr:
        """
        Retrieve historical data on the returns any of the entities supported have recorded.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> apr = eth.operator.apr("Lido", id_type=IdType.POOL, time_window=TimeWindow.ALL_TIME)
            >>> print(f"{apr.percentage = }%")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class
            time_window: The time window of aggregation
            apr_type: Direction of flow

        Returns:
            Entity APR %
        """
        url: str = f"{self.resource_path}/{operator_id}/apr"
        params: Dict[str, Any] = {
            "idType": id_type.value,
            "window": time_window.value,
            "aprType": apr_type.value,
        }
        data = self.client.get(url, params=params)
        return json_to_instance(data, OperatorApr)

    def summary(
        self,
        operator_id: str,
        id_type: IdType,
        *,
        time_window: TimeWindow,
    ) -> OperatorSummary:
        """
        Retrieve summary statistics for a specific operator

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> summary = eth.operator.summary("Lido", id_type=IdType.POOL, time_window=TimeWindow.SEVEN_DAYS)
            >>> print(f"{summary.avg_uptime = }%")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class
            time_window: The time window of aggregation

        Returns:
            Operator summary
        """
        url: str = f"{self.resource_path}/{operator_id}/summary"
        params: Dict[str, Any] = {
            "idType": id_type.value,
            "window": time_window.value,
        }
        data = self.client.get(url, params=params)
        return json_to_instance(data, OperatorSummary)

    def stake_movement(
        self,
        operator_id: str,
        id_type: IdType,
        *,
        stake_action: StakeAction = StakeAction.ACTIVATION,
        time_window: TimeWindow,
    ) -> Iterator[OperatorStakeMovement]:
        """
        Retrieve data on the activation and exit activity of a specific pre-materialized view
        (e.g. operator, deposit address, etc.)

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> stake_movement = eth.operator.stake_movement("Lido", id_type=IdType.POOL, time_window=TimeWindow.THIRTY_DAYS)
            >>> for mov in stake_movement:
            >>>     print(f"{mov.amount_gwei = }")

        Args:
            operator_id: The name of the entity in question
            id_type: The type of entity class
            stake_action: Direction of flow
            time_window: The time window of aggregation

        Yields:
            Activations and withdrawals state and status

        """
        url: str = f"{self.resource_path}/{operator_id}/stakeMovement"
        params: Dict[str, Any] = {
            "idType": id_type.value,
            "stakeAction": stake_action.value,
            "window": time_window.value,
        }
        data = self.client.get(url, params=params)
        for item in data:
            yield json_to_instance(item, OperatorStakeMovement)

apr(operator_id, id_type, *, time_window, apr_type=AprType.BACKWARD)

Retrieve historical data on the returns any of the entities supported have recorded.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> apr = eth.operator.apr("Lido", id_type=IdType.POOL, time_window=TimeWindow.ALL_TIME)
>>> print(f"{apr.percentage = }%")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required
time_window TimeWindow

The time window of aggregation

 required
apr_type AprType

Direction of flow

 BACKWARD

Returns:

Type Description
OperatorApr

Entity APR %
Source code in src/rated/ethereum/operators.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def apr(
    self,
    operator_id: str,
    id_type: IdType,
    *,
    time_window: TimeWindow,
    apr_type: AprType = AprType.BACKWARD,
) -> OperatorApr:
    """
    Retrieve historical data on the returns any of the entities supported have recorded.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> apr = eth.operator.apr("Lido", id_type=IdType.POOL, time_window=TimeWindow.ALL_TIME)
        >>> print(f"{apr.percentage = }%")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class
        time_window: The time window of aggregation
        apr_type: Direction of flow

    Returns:
        Entity APR %
    """
    url: str = f"{self.resource_path}/{operator_id}/apr"
    params: Dict[str, Any] = {
        "idType": id_type.value,
        "window": time_window.value,
        "aprType": apr_type.value,
    }
    data = self.client.get(url, params=params)
    return json_to_instance(data, OperatorApr)

clients(operator_id, id_type)

Consensus client distribution

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> clients = eth.operator.clients("Lido", id_type=IdType.POOL)
>>> for c in clients:
>>>     print(f"{c.client = }, {c.percentage = }%")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required

Yields:

Type Description
ClientPercentage

Clients percentages
Source code in src/rated/ethereum/operators.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def clients(self, operator_id: str, id_type: IdType) -> Iterator[ClientPercentage]:
    """
    Consensus client distribution

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> clients = eth.operator.clients("Lido", id_type=IdType.POOL)
        >>> for c in clients:
        >>>     print(f"{c.client = }, {c.percentage = }%")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class

    Yields:
        Clients percentages
    """
    url: str = f"{self.resource_path}/{operator_id}/clients"
    params: Dict[str, Any] = {"idType": id_type.value}
    data = self.client.get(url, params=params)
    for item in data:
        yield json_to_instance(item, ClientPercentage)

effectiveness(operator_id, id_type, *, from_day=None, size=None, granularity=Granularity.DAY, filter_type=FilterType.DAY, follow_next=False)

Historical performance of a single operator. This includes rewards (aggregate and granular), performance (effectiveness and its components), slashing history and much more.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> effectiveness = eth.operator.effectiveness("Lido", id_type=IdType.POOL, from_day=795, size=10)
>>> for eff in effectiveness:
>>>     print(f"{eff.avg_validator_effectiveness = }, {eff.day = }")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required
from_day int | date | None

Start day

 None
size int | None

Number of results included per page

 None
granularity Granularity

T he size of time increments you are looking to query

 DAY
filter_type FilterType

Hour, day and datetime

 DAY
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
OperatorEffectiveness

Operator Effectiveness
Source code in src/rated/ethereum/operators.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def effectiveness(
    self,
    operator_id: str,
    id_type: IdType,
    *,
    from_day: int | date | None = None,
    size: int | None = None,
    granularity: Granularity = Granularity.DAY,
    filter_type: FilterType = FilterType.DAY,
    follow_next: bool = False,
) -> Iterator[OperatorEffectiveness]:
    """
    Historical performance of a single operator.
    This includes rewards (aggregate and granular), performance (effectiveness and its components),
    slashing history and much more.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> effectiveness = eth.operator.effectiveness("Lido", id_type=IdType.POOL, from_day=795, size=10)
        >>> for eff in effectiveness:
        >>>     print(f"{eff.avg_validator_effectiveness = }, {eff.day = }")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class
        from_day: Start day
        size: Number of results included per page
        granularity:T he size of time increments you are looking to query
        filter_type: Hour, day and datetime
        follow_next: Whether to follow pagination or not

    Yields:
        Operator Effectiveness
    """
    url: str = f"{self.resource_path}/{operator_id}/effectiveness"
    params: Dict[str, Any] = {
        "idType": id_type.value,
        "from": from_day,
        "size": size,
        "granularity": granularity.value,
        "filterType": filter_type.value,
    }
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=OperatorEffectiveness,
        follow_next=follow_next,
    )

metadata(operator_id, id_type)

Retrieve profile information on specific operators.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> op = eth.operator.metadata("Lido", id_type=IdType.POOL)
>>> print(f"{op.node_operator_count = }")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required

Returns:

Type Description
Operator

Operator metadata.
Source code in src/rated/ethereum/operators.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def metadata(self, operator_id: str, id_type: IdType) -> OperatorType:
    """
    Retrieve profile information on specific operators.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> op = eth.operator.metadata("Lido", id_type=IdType.POOL)
        >>> print(f"{op.node_operator_count = }")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class

    Returns:
        Operator metadata.
    """
    url: str = f"{self.resource_path}/{operator_id}"
    params: Dict[str, Any] = {"idType": id_type.value}
    operator = self.client.get(url, params=params)
    return json_to_instance(operator, OperatorType)

relayers(operator_id, id_type, *, time_window=TimeWindow.THIRTY_DAYS)

Get information relating to an entity's historical distribution of relays they have procured blocks from.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> relayers = eth.operator.relayers("Lido", id_type=IdType.POOL, time_window=TimeWindow.ALL_TIME)
>>> for r in relayers:
>>>     print(f"{r.relayer = }, {r.percentage = }%")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required
time_window TimeWindow

The time window of aggregation

 THIRTY_DAYS

Yields:

Type Description
RelayerPercentage

Relayer Percentages
Source code in src/rated/ethereum/operators.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def relayers(
    self,
    operator_id: str,
    id_type: IdType,
    *,
    time_window: TimeWindow = TimeWindow.THIRTY_DAYS,
) -> Iterator[RelayerPercentage]:
    """
    Get information relating to an entity's historical distribution of relays they have procured blocks from.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> relayers = eth.operator.relayers("Lido", id_type=IdType.POOL, time_window=TimeWindow.ALL_TIME)
        >>> for r in relayers:
        >>>     print(f"{r.relayer = }, {r.percentage = }%")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class
        time_window: The time window of aggregation

    Yields:
        Relayer Percentages
    """
    url: str = f"{self.resource_path}/{operator_id}/relayers"
    params: Dict[str, Any] = {"idType": id_type.value, "window": time_window.value}
    data = self.client.get(url, params=params)
    for item in data:
        yield json_to_instance(item, RelayerPercentage)

stake_movement(operator_id, id_type, *, stake_action=StakeAction.ACTIVATION, time_window)

Retrieve data on the activation and exit activity of a specific pre-materialized view (e.g. operator, deposit address, etc.)

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> stake_movement = eth.operator.stake_movement("Lido", id_type=IdType.POOL, time_window=TimeWindow.THIRTY_DAYS)
>>> for mov in stake_movement:
>>>     print(f"{mov.amount_gwei = }")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required
stake_action StakeAction

Direction of flow

 ACTIVATION
time_window TimeWindow

The time window of aggregation

 required

Yields:

Type Description
OperatorStakeMovement

Activations and withdrawals state and status
Source code in src/rated/ethereum/operators.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def stake_movement(
    self,
    operator_id: str,
    id_type: IdType,
    *,
    stake_action: StakeAction = StakeAction.ACTIVATION,
    time_window: TimeWindow,
) -> Iterator[OperatorStakeMovement]:
    """
    Retrieve data on the activation and exit activity of a specific pre-materialized view
    (e.g. operator, deposit address, etc.)

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> stake_movement = eth.operator.stake_movement("Lido", id_type=IdType.POOL, time_window=TimeWindow.THIRTY_DAYS)
        >>> for mov in stake_movement:
        >>>     print(f"{mov.amount_gwei = }")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class
        stake_action: Direction of flow
        time_window: The time window of aggregation

    Yields:
        Activations and withdrawals state and status

    """
    url: str = f"{self.resource_path}/{operator_id}/stakeMovement"
    params: Dict[str, Any] = {
        "idType": id_type.value,
        "stakeAction": stake_action.value,
        "window": time_window.value,
    }
    data = self.client.get(url, params=params)
    for item in data:
        yield json_to_instance(item, OperatorStakeMovement)

summary(operator_id, id_type, *, time_window)

Retrieve summary statistics for a specific operator

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> summary = eth.operator.summary("Lido", id_type=IdType.POOL, time_window=TimeWindow.SEVEN_DAYS)
>>> print(f"{summary.avg_uptime = }%")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
id_type IdType

The type of entity class

 required
time_window TimeWindow

The time window of aggregation

 required

Returns:

Type Description
OperatorSummary

Operator summary
Source code in src/rated/ethereum/operators.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def summary(
    self,
    operator_id: str,
    id_type: IdType,
    *,
    time_window: TimeWindow,
) -> OperatorSummary:
    """
    Retrieve summary statistics for a specific operator

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> summary = eth.operator.summary("Lido", id_type=IdType.POOL, time_window=TimeWindow.SEVEN_DAYS)
        >>> print(f"{summary.avg_uptime = }%")

    Args:
        operator_id: The name of the entity in question
        id_type: The type of entity class
        time_window: The time window of aggregation

    Returns:
        Operator summary
    """
    url: str = f"{self.resource_path}/{operator_id}/summary"
    params: Dict[str, Any] = {
        "idType": id_type.value,
        "window": time_window.value,
    }
    data = self.client.get(url, params=params)
    return json_to_instance(data, OperatorSummary)

Operators

Bases: APIResource

Source code in src/rated/ethereum/operators.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
class Operators(APIResource):
    path = "/operators"

    def percentiles(
        self,
        id_type: IdType,
        *,
        time_window: TimeWindow,
    ) -> Iterator[Percentile]:
        """
        Retrieve data of entities with their respective percentile rank score, according to their effectiveness rating.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> percentiles = eth.operators.percentiles(IdType.POOL, time_window=TimeWindow.SEVEN_DAYS)
            >>> for percentile in percentiles:
            >>>     print(f"{percentile.rank = }, {percentile.value = }")

        Args:
            id_type: The type of entity class
            time_window: The time window of aggregation

        Yields:
            Percentiles

        See Also:
            https://docs.rated.network/methodologies/ethereum-beacon-chain/rating-percentiles

        """
        url: str = f"{self.resource_path}/percentiles"
        params: Dict[str, Any] = {
            "idType": id_type.value,
            "window": time_window.value,
        }
        data = self.client.get(url, params=params)
        for item in data:
            yield json_to_instance(item, Percentile)

    def summaries(
        self,
        *,
        time_window: TimeWindow,
        pool_type: PoolType = PoolType.ALL,
        id_type: IdType = IdType.DEPOSIT_ADDRESS,
        parent_id: str | None = None,
        from_day: int | None = None,
        size: int | None = None,
        follow_next: bool = False,
    ) -> Iterator[OperatorSummary]:
        """
        Summarizes statistics for all the operators Rated has pre-materialized views on

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> summaries = eth.operators.summaries(time_window=TimeWindow.ALL_TIME, from_day=795, size=10)
            >>> for summary in summaries:
            >>>     print(f"{summary.avg_uptime = }")

        Args:
            time_window: The time window of aggregation
            pool_type: Type of Pool
            id_type: The type of entity class
            parent_id: Specifying a pool or node operator so that the response is focused on the Pool Shares of said entity
            from_day: Start day
            size: Number of results included per page
            follow_next: Whether to follow pagination or not

        Yields:

        """
        url: str = f"{self.resource_path}"
        params: Dict[str, Any] = {
            "poolType": pool_type.value,
            "idType": id_type.value,
            "parentId": parent_id,
            "window": time_window.value,
            "from": from_day,
            "size": size,
        }
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=OperatorSummary,
            follow_next=follow_next,
        )

percentiles(id_type, *, time_window)

Retrieve data of entities with their respective percentile rank score, according to their effectiveness rating.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> percentiles = eth.operators.percentiles(IdType.POOL, time_window=TimeWindow.SEVEN_DAYS)
>>> for percentile in percentiles:
>>>     print(f"{percentile.rank = }, {percentile.value = }")

Parameters:

Name Type Description Default
id_type IdType

The type of entity class

 required
time_window TimeWindow

The time window of aggregation

 required

Yields:

Type Description
Percentile

Percentiles
See Also

https://docs.rated.network/methodologies/ethereum-beacon-chain/rating-percentiles

Source code in src/rated/ethereum/operators.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def percentiles(
    self,
    id_type: IdType,
    *,
    time_window: TimeWindow,
) -> Iterator[Percentile]:
    """
    Retrieve data of entities with their respective percentile rank score, according to their effectiveness rating.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> percentiles = eth.operators.percentiles(IdType.POOL, time_window=TimeWindow.SEVEN_DAYS)
        >>> for percentile in percentiles:
        >>>     print(f"{percentile.rank = }, {percentile.value = }")

    Args:
        id_type: The type of entity class
        time_window: The time window of aggregation

    Yields:
        Percentiles

    See Also:
        https://docs.rated.network/methodologies/ethereum-beacon-chain/rating-percentiles

    """
    url: str = f"{self.resource_path}/percentiles"
    params: Dict[str, Any] = {
        "idType": id_type.value,
        "window": time_window.value,
    }
    data = self.client.get(url, params=params)
    for item in data:
        yield json_to_instance(item, Percentile)

summaries(*, time_window, pool_type=PoolType.ALL, id_type=IdType.DEPOSIT_ADDRESS, parent_id=None, from_day=None, size=None, follow_next=False)

Summarizes statistics for all the operators Rated has pre-materialized views on

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> summaries = eth.operators.summaries(time_window=TimeWindow.ALL_TIME, from_day=795, size=10)
>>> for summary in summaries:
>>>     print(f"{summary.avg_uptime = }")

Parameters:

Name Type Description Default
time_window TimeWindow

The time window of aggregation

 required
pool_type PoolType

Type of Pool

 ALL
id_type IdType

The type of entity class

 DEPOSIT_ADDRESS
parent_id str | None

Specifying a pool or node operator so that the response is focused on the Pool Shares of said entity

 None
from_day int | None

Start day

 None
size int | None

Number of results included per page

 None
follow_next bool

Whether to follow pagination or not

 False

Yields:

Source code in src/rated/ethereum/operators.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
def summaries(
    self,
    *,
    time_window: TimeWindow,
    pool_type: PoolType = PoolType.ALL,
    id_type: IdType = IdType.DEPOSIT_ADDRESS,
    parent_id: str | None = None,
    from_day: int | None = None,
    size: int | None = None,
    follow_next: bool = False,
) -> Iterator[OperatorSummary]:
    """
    Summarizes statistics for all the operators Rated has pre-materialized views on

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> summaries = eth.operators.summaries(time_window=TimeWindow.ALL_TIME, from_day=795, size=10)
        >>> for summary in summaries:
        >>>     print(f"{summary.avg_uptime = }")

    Args:
        time_window: The time window of aggregation
        pool_type: Type of Pool
        id_type: The type of entity class
        parent_id: Specifying a pool or node operator so that the response is focused on the Pool Shares of said entity
        from_day: Start day
        size: Number of results included per page
        follow_next: Whether to follow pagination or not

    Yields:

    """
    url: str = f"{self.resource_path}"
    params: Dict[str, Any] = {
        "poolType": pool_type.value,
        "idType": id_type.value,
        "parentId": parent_id,
        "window": time_window.value,
        "from": from_day,
        "size": size,
    }
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=OperatorSummary,
        follow_next=follow_next,
    )

P2P

Bases: APIResource

Querying into aggregated stats on Ethereum's peer-to-peer networking layer

Source code in src/rated/ethereum/p2p.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class P2P(APIResource):
    """
    Querying into aggregated stats on Ethereum's peer-to-peer networking layer
    """

    path = "/p2p"

    def geographical_distribution(
        self,
        distribution_type: DistributionType = DistributionType.PROS,
    ) -> Iterator[P2PGeographicalDistribution]:
        """
        Retrieves a list of countries and the respective share of the validator set based in those countries

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> geo_dist = eth.p2p.geographical_distribution()
            >>> for dist in geo_dist:
            >>>     print(f"{dist.country = }, {dist.validator_share = }")

        Args:
            distribution_type: The type of distribution

        Yields:
            Geographical distribution
        """
        url = f"{self.resource_path}/geographical"
        params = {"distType": distribution_type.value}
        data = self.client.get(url, params=params)
        for item in data:
            yield json_to_instance(item, P2PGeographicalDistribution)

    def hosting_provider_distribution(
        self,
        *,
        from_rank: int | None = None,
        size: int | None = None,
        distribution_type: DistributionType = DistributionType.PROS,
        follow_next: bool = False,
    ) -> Iterator[P2PHostingProviderDistribution]:
        """
        Retrieves a list of hosting providers and their respective share of the validator set

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> provider_dist = eth.p2p.hosting_provider_distribution(size=5)
            >>> for dist in provider_dist:
            >>>     print(f"{dist.hosting_provider = }, {dist.validator_share = }")

        Args:
            from_rank:
            size: Number of results included per page
            distribution_type: The type of distribution
            follow_next: Whether to follow pagination or not

        Yields:
            Hosting provider distribution
        """
        url = f"{self.resource_path}/hostingProvider"
        params = {"from": from_rank, "size": size, "distType": distribution_type.value}
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=P2PHostingProviderDistribution,
            follow_next=follow_next,
        )

geographical_distribution(distribution_type=DistributionType.PROS)

Retrieves a list of countries and the respective share of the validator set based in those countries

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> geo_dist = eth.p2p.geographical_distribution()
>>> for dist in geo_dist:
>>>     print(f"{dist.country = }, {dist.validator_share = }")

Parameters:

Name Type Description Default
distribution_type DistributionType

The type of distribution

 PROS

Yields:

Type Description
P2PGeographicalDistribution

Geographical distribution
Source code in src/rated/ethereum/p2p.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def geographical_distribution(
    self,
    distribution_type: DistributionType = DistributionType.PROS,
) -> Iterator[P2PGeographicalDistribution]:
    """
    Retrieves a list of countries and the respective share of the validator set based in those countries

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> geo_dist = eth.p2p.geographical_distribution()
        >>> for dist in geo_dist:
        >>>     print(f"{dist.country = }, {dist.validator_share = }")

    Args:
        distribution_type: The type of distribution

    Yields:
        Geographical distribution
    """
    url = f"{self.resource_path}/geographical"
    params = {"distType": distribution_type.value}
    data = self.client.get(url, params=params)
    for item in data:
        yield json_to_instance(item, P2PGeographicalDistribution)

hosting_provider_distribution(*, from_rank=None, size=None, distribution_type=DistributionType.PROS, follow_next=False)

Retrieves a list of hosting providers and their respective share of the validator set

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> provider_dist = eth.p2p.hosting_provider_distribution(size=5)
>>> for dist in provider_dist:
>>>     print(f"{dist.hosting_provider = }, {dist.validator_share = }")

Parameters:

Name Type Description Default
from_rank int | None
None
size int | None

Number of results included per page

 None
distribution_type DistributionType

The type of distribution

 PROS
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
P2PHostingProviderDistribution

Hosting provider distribution
Source code in src/rated/ethereum/p2p.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def hosting_provider_distribution(
    self,
    *,
    from_rank: int | None = None,
    size: int | None = None,
    distribution_type: DistributionType = DistributionType.PROS,
    follow_next: bool = False,
) -> Iterator[P2PHostingProviderDistribution]:
    """
    Retrieves a list of hosting providers and their respective share of the validator set

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> provider_dist = eth.p2p.hosting_provider_distribution(size=5)
        >>> for dist in provider_dist:
        >>>     print(f"{dist.hosting_provider = }, {dist.validator_share = }")

    Args:
        from_rank:
        size: Number of results included per page
        distribution_type: The type of distribution
        follow_next: Whether to follow pagination or not

    Yields:
        Hosting provider distribution
    """
    url = f"{self.resource_path}/hostingProvider"
    params = {"from": from_rank, "size": size, "distType": distribution_type.value}
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=P2PHostingProviderDistribution,
        follow_next=follow_next,
    )

Slashings

Bases: APIResource

Allows one to see every slashed validator in the Ethereum Beacon Chain whether individually or collectively. Pertinent metrics include their total penalties from being slashed, which epoch they were slashed, and when they will be withdrawable.

Source code in src/rated/ethereum/slashings.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class Slashings(APIResource):
    """
    Allows one to see every slashed validator in the Ethereum Beacon Chain whether individually or collectively.
    Pertinent metrics include their total penalties from being slashed,
    which epoch they were slashed, and when they will be withdrawable.
    """

    path = "/slashings"

    def overview(self) -> Iterator[SlashingOverview]:
        """
        Lists of all slashed validators, their index, pubkey, slashing epoch, withdrawable epoch,
        balance before slashing, balance before withdrawal, and the penalties incurred from getting slashed.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> slashings_overview = eth.slashings.overview()
            >>> for slashing in slashings_overview:
            >>>     print(f"{slashing.validators_slashed = }")

        Yields:
            Slashing overview
        """
        url: str = f"{self.resource_path}/overview"
        data = self.client.get(url)
        for item in data:
            yield json_to_instance(item, SlashingOverview)

    def leaderboard(
        self,
        *,
        from_rank: int | None = None,
        size: int | None = None,
        follow_next: bool = False,
    ) -> Iterator[SlashingLeaderboard]:
        """
        Depending on the slashing role specified, this endpoint returns a list of entities either
        (1) according to how many times their validators have been slashed or
        (2) how many times their validators have proposed a block that included slashing report
           (i.e. letting the network know a slashing incident has occurred)

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> leaderboard = eth.slashings.leaderboard(from_rank=1)
            >>> for l in leaderboard:
            >>>     print(f"{l.id = }, {l.slashes = }, {l.validator_count = }")

        Args:
            from_rank: Start from ranking
            size: Number of results included per page
            follow_next: Whether to follow pagination or not

        Yields:
            Slashing leaderboard
        """
        url: str = f"{self.resource_path}/leaderboard"
        params = {"from": from_rank, "size": size}
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=SlashingLeaderboard,
            follow_next=follow_next,
        )

    def cohorts(self) -> Iterator[SlashingCohort]:
        """
        Retrieves the frequency of slashing incidents for validators, grouped by different operator cohort sizes,
        from solo to professional operators with more than 5,000 validator keys.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> cohorts = eth.slashings.cohorts()
            >>> for cohort in cohorts:
            >>>     print(f"{cohort.cohort = }, {cohort.last_six_months = }")

        Yield:
            Cohorts
        """
        url: str = f"{self.resource_path}/cohortAnalysis"
        data = self.client.get(url)
        for item in data:
            yield json_to_instance(item, SlashingCohort)

    def timeseries(self) -> Iterator[SlashingTimeInterval]:
        """
        Time series of slashing incidents

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> intervals = eth.slashings.timeseries()
            >>> for interval in intervals:
            >>>     print(f"{interval.month = }, {interval.validators_slashed = }")

        Yields:
            Slashing time series
        """
        url: str = f"{self.resource_path}/timeseries"
        data = self.client.get(url)
        for item in data:
            yield json_to_instance(item, SlashingTimeInterval)

    def penalties(
        self,
        *,
        from_day: int | date | None = None,
        size: int | None = None,
        follow_next: bool = False,
    ) -> Iterator[SlashingPenalty]:
        """
        All slashed validators, their index, pubkey, slashing epoch, withdrawable epoch, balance before slashing,
        balance before withdrawal, and the penalties incurred from getting slashed

        Examples:
            >>> from datetime import date
            >>>
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> penalties = eth.slashings.penalties(from_day=date(2023, 10, 1), size=10)
            >>> for penalty in penalties:
            >>>     print(f"{penalty.validator_pubkey = }, {penalty.slashing_penalties = }")

        Args:
            from_day: Starting day
            size: Number of results included per page
            follow_next: Whether to follow pagination or not

        Yields:
            Slashing penalty

        """
        url: str = f"{self.resource_path}"
        from_: str | int | date | None = from_day
        if from_ is not None and isinstance(from_, date):
            from_ = from_.isoformat()
        params: Dict[str, Any] = {"from": from_, "size": size}
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=SlashingPenalty,
            follow_next=follow_next,
        )

    def for_validator(
        self,
        validator_index_or_pubkey: int | str,
    ) -> SlashingPenalty:
        """
        Information about a single slashed validator, queried either by the validator's index or their pubkey.

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> penalties = eth.slashings.for_validator("0xb443c10134d35b2f2117b0cd499f0e1755a87329d573a680cec6830fa2f7032f8dc39be33dc75b7a83c2ae70651eb38b")
            >>> print(f"{penalties.slashing_epoch = }, {penalties.slashing_penalties = }")

        Args:
            validator_index_or_pubkey: Validator index or pubkey

        Returns:
            Slashing penalty
        """
        url: str = f"{self.resource_path}/{validator_index_or_pubkey}"
        data = self.client.get(url)
        return json_to_instance(data, SlashingPenalty)

cohorts()

Retrieves the frequency of slashing incidents for validators, grouped by different operator cohort sizes, from solo to professional operators with more than 5,000 validator keys.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> cohorts = eth.slashings.cohorts()
>>> for cohort in cohorts:
>>>     print(f"{cohort.cohort = }, {cohort.last_six_months = }")
Yield

Cohorts

Source code in src/rated/ethereum/slashings.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def cohorts(self) -> Iterator[SlashingCohort]:
    """
    Retrieves the frequency of slashing incidents for validators, grouped by different operator cohort sizes,
    from solo to professional operators with more than 5,000 validator keys.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> cohorts = eth.slashings.cohorts()
        >>> for cohort in cohorts:
        >>>     print(f"{cohort.cohort = }, {cohort.last_six_months = }")

    Yield:
        Cohorts
    """
    url: str = f"{self.resource_path}/cohortAnalysis"
    data = self.client.get(url)
    for item in data:
        yield json_to_instance(item, SlashingCohort)

for_validator(validator_index_or_pubkey)

Information about a single slashed validator, queried either by the validator's index or their pubkey.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> penalties = eth.slashings.for_validator("0xb443c10134d35b2f2117b0cd499f0e1755a87329d573a680cec6830fa2f7032f8dc39be33dc75b7a83c2ae70651eb38b")
>>> print(f"{penalties.slashing_epoch = }, {penalties.slashing_penalties = }")

Parameters:

Name Type Description Default
validator_index_or_pubkey int | str

Validator index or pubkey

 required

Returns:

Type Description
SlashingPenalty

Slashing penalty
Source code in src/rated/ethereum/slashings.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def for_validator(
    self,
    validator_index_or_pubkey: int | str,
) -> SlashingPenalty:
    """
    Information about a single slashed validator, queried either by the validator's index or their pubkey.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> penalties = eth.slashings.for_validator("0xb443c10134d35b2f2117b0cd499f0e1755a87329d573a680cec6830fa2f7032f8dc39be33dc75b7a83c2ae70651eb38b")
        >>> print(f"{penalties.slashing_epoch = }, {penalties.slashing_penalties = }")

    Args:
        validator_index_or_pubkey: Validator index or pubkey

    Returns:
        Slashing penalty
    """
    url: str = f"{self.resource_path}/{validator_index_or_pubkey}"
    data = self.client.get(url)
    return json_to_instance(data, SlashingPenalty)

leaderboard(*, from_rank=None, size=None, follow_next=False)

Depending on the slashing role specified, this endpoint returns a list of entities either (1) according to how many times their validators have been slashed or (2) how many times their validators have proposed a block that included slashing report (i.e. letting the network know a slashing incident has occurred)

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> leaderboard = eth.slashings.leaderboard(from_rank=1)
>>> for l in leaderboard:
>>>     print(f"{l.id = }, {l.slashes = }, {l.validator_count = }")

Parameters:

Name Type Description Default
from_rank int | None

Start from ranking

 None
size int | None

Number of results included per page

 None
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
SlashingLeaderboard

Slashing leaderboard
Source code in src/rated/ethereum/slashings.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def leaderboard(
    self,
    *,
    from_rank: int | None = None,
    size: int | None = None,
    follow_next: bool = False,
) -> Iterator[SlashingLeaderboard]:
    """
    Depending on the slashing role specified, this endpoint returns a list of entities either
    (1) according to how many times their validators have been slashed or
    (2) how many times their validators have proposed a block that included slashing report
       (i.e. letting the network know a slashing incident has occurred)

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> leaderboard = eth.slashings.leaderboard(from_rank=1)
        >>> for l in leaderboard:
        >>>     print(f"{l.id = }, {l.slashes = }, {l.validator_count = }")

    Args:
        from_rank: Start from ranking
        size: Number of results included per page
        follow_next: Whether to follow pagination or not

    Yields:
        Slashing leaderboard
    """
    url: str = f"{self.resource_path}/leaderboard"
    params = {"from": from_rank, "size": size}
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=SlashingLeaderboard,
        follow_next=follow_next,
    )

overview()

Lists of all slashed validators, their index, pubkey, slashing epoch, withdrawable epoch, balance before slashing, balance before withdrawal, and the penalties incurred from getting slashed.

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> slashings_overview = eth.slashings.overview()
>>> for slashing in slashings_overview:
>>>     print(f"{slashing.validators_slashed = }")

Yields:

Type Description
SlashingOverview

Slashing overview
Source code in src/rated/ethereum/slashings.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def overview(self) -> Iterator[SlashingOverview]:
    """
    Lists of all slashed validators, their index, pubkey, slashing epoch, withdrawable epoch,
    balance before slashing, balance before withdrawal, and the penalties incurred from getting slashed.

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> slashings_overview = eth.slashings.overview()
        >>> for slashing in slashings_overview:
        >>>     print(f"{slashing.validators_slashed = }")

    Yields:
        Slashing overview
    """
    url: str = f"{self.resource_path}/overview"
    data = self.client.get(url)
    for item in data:
        yield json_to_instance(item, SlashingOverview)

penalties(*, from_day=None, size=None, follow_next=False)

All slashed validators, their index, pubkey, slashing epoch, withdrawable epoch, balance before slashing, balance before withdrawal, and the penalties incurred from getting slashed

Examples:

>>> from datetime import date
>>>
>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> penalties = eth.slashings.penalties(from_day=date(2023, 10, 1), size=10)
>>> for penalty in penalties:
>>>     print(f"{penalty.validator_pubkey = }, {penalty.slashing_penalties = }")

Parameters:

Name Type Description Default
from_day int | date | None

Starting day

 None
size int | None

Number of results included per page

 None
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
SlashingPenalty

Slashing penalty
Source code in src/rated/ethereum/slashings.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def penalties(
    self,
    *,
    from_day: int | date | None = None,
    size: int | None = None,
    follow_next: bool = False,
) -> Iterator[SlashingPenalty]:
    """
    All slashed validators, their index, pubkey, slashing epoch, withdrawable epoch, balance before slashing,
    balance before withdrawal, and the penalties incurred from getting slashed

    Examples:
        >>> from datetime import date
        >>>
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> penalties = eth.slashings.penalties(from_day=date(2023, 10, 1), size=10)
        >>> for penalty in penalties:
        >>>     print(f"{penalty.validator_pubkey = }, {penalty.slashing_penalties = }")

    Args:
        from_day: Starting day
        size: Number of results included per page
        follow_next: Whether to follow pagination or not

    Yields:
        Slashing penalty

    """
    url: str = f"{self.resource_path}"
    from_: str | int | date | None = from_day
    if from_ is not None and isinstance(from_, date):
        from_ = from_.isoformat()
    params: Dict[str, Any] = {"from": from_, "size": size}
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=SlashingPenalty,
        follow_next=follow_next,
    )

timeseries()

Time series of slashing incidents

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> intervals = eth.slashings.timeseries()
>>> for interval in intervals:
>>>     print(f"{interval.month = }, {interval.validators_slashed = }")

Yields:

Type Description
SlashingTimeInterval

Slashing time series
Source code in src/rated/ethereum/slashings.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def timeseries(self) -> Iterator[SlashingTimeInterval]:
    """
    Time series of slashing incidents

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> intervals = eth.slashings.timeseries()
        >>> for interval in intervals:
        >>>     print(f"{interval.month = }, {interval.validators_slashed = }")

    Yields:
        Slashing time series
    """
    url: str = f"{self.resource_path}/timeseries"
    data = self.client.get(url)
    for item in data:
        yield json_to_instance(item, SlashingTimeInterval)

Validator

Bases: APIResource

Source code in src/rated/ethereum/validators.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class Validator(APIResource):
    path = "/validators"

    def metadata(self, index_or_pubkey: int | str) -> ValidatorMetadata:
        """
        Reverse lookup into the entity-to-validator index mappings that live in the RatedDB

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> metadata = eth.validator.metadata(560000)
            >>> print(f"{metadata.validator_pubkey = }, {metadata.deposit_addresses = }")

        Args:
            index_or_pubkey: Validator index or pubkey

        Returns:
            Metadata about the validator
        """
        validator = self.client.get(f"{self.resource_path}/{index_or_pubkey}")
        return json_to_instance(validator, ValidatorMetadata)

    def apr(
        self,
        index_or_pubkey: int | str,
        *,
        apr_type: AprType = AprType.BACKWARD,
        time_window: TimeWindow = TimeWindow.ONE_DAY,
    ) -> ValidatorAPR:
        """
        Historical data on the returns of a validator index

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> validator_apr = eth.validator.apr(560000, time_window=TimeWindow.THIRTY_DAYS)
            >>> print(f"{validator_apr.validator_index = }, {validator_apr.percentage = }%")

        Args:
            index_or_pubkey: Validator index or pubkey
            apr_type: Direction of flow
            time_window: The time window of aggregation

        Returns:
            APR %
        """
        params = {"aprType": apr_type.value, "window": time_window.value}
        apr = self.client.get(
            f"{self.resource_path}/{index_or_pubkey}/apr",
            params=params,
        )
        return json_to_instance(apr, ValidatorAPR)

    def effectiveness(
        self,
        index_or_pubkey: int | str,
        *,
        from_day: int | date | None = None,
        size: int | None = None,
        follow_next: bool = False,
    ) -> Iterator[ValidatorEffectiveness]:
        """
        Historical performance of a single validator index

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> effectiveness = eth.validator.effectiveness(560000, from_day=795)
            >>> for eff in effectiveness:
            >>>     print(f"{eff.validator_index = }, {eff.validator_effectiveness = }")

        Args:
            index_or_pubkey: Validator index or pubkey
            from_day: Starting day
            size: Number of results included per page
            follow_next: Whether to follow pagination or not

        Yields:
            Effectiveness metrics
        """
        from_: str | int | date | None = from_day
        if from_ is not None and isinstance(from_, date):
            from_ = from_.isoformat()

        params: Dict[str, Any] = {"from": from_, "size": size}
        url = f"{self.resource_path}/{index_or_pubkey}/effectiveness"
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=ValidatorEffectiveness,
            follow_next=follow_next,
        )

apr(index_or_pubkey, *, apr_type=AprType.BACKWARD, time_window=TimeWindow.ONE_DAY)

Historical data on the returns of a validator index

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> validator_apr = eth.validator.apr(560000, time_window=TimeWindow.THIRTY_DAYS)
>>> print(f"{validator_apr.validator_index = }, {validator_apr.percentage = }%")

Parameters:

Name Type Description Default
index_or_pubkey int | str

Validator index or pubkey

 required
apr_type AprType

Direction of flow

 BACKWARD
time_window TimeWindow

The time window of aggregation

 ONE_DAY

Returns:

Type Description
ValidatorAPR

APR %
Source code in src/rated/ethereum/validators.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def apr(
    self,
    index_or_pubkey: int | str,
    *,
    apr_type: AprType = AprType.BACKWARD,
    time_window: TimeWindow = TimeWindow.ONE_DAY,
) -> ValidatorAPR:
    """
    Historical data on the returns of a validator index

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> validator_apr = eth.validator.apr(560000, time_window=TimeWindow.THIRTY_DAYS)
        >>> print(f"{validator_apr.validator_index = }, {validator_apr.percentage = }%")

    Args:
        index_or_pubkey: Validator index or pubkey
        apr_type: Direction of flow
        time_window: The time window of aggregation

    Returns:
        APR %
    """
    params = {"aprType": apr_type.value, "window": time_window.value}
    apr = self.client.get(
        f"{self.resource_path}/{index_or_pubkey}/apr",
        params=params,
    )
    return json_to_instance(apr, ValidatorAPR)

effectiveness(index_or_pubkey, *, from_day=None, size=None, follow_next=False)

Historical performance of a single validator index

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> effectiveness = eth.validator.effectiveness(560000, from_day=795)
>>> for eff in effectiveness:
>>>     print(f"{eff.validator_index = }, {eff.validator_effectiveness = }")

Parameters:

Name Type Description Default
index_or_pubkey int | str

Validator index or pubkey

 required
from_day int | date | None

Starting day

 None
size int | None

Number of results included per page

 None
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
ValidatorEffectiveness

Effectiveness metrics
Source code in src/rated/ethereum/validators.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def effectiveness(
    self,
    index_or_pubkey: int | str,
    *,
    from_day: int | date | None = None,
    size: int | None = None,
    follow_next: bool = False,
) -> Iterator[ValidatorEffectiveness]:
    """
    Historical performance of a single validator index

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> effectiveness = eth.validator.effectiveness(560000, from_day=795)
        >>> for eff in effectiveness:
        >>>     print(f"{eff.validator_index = }, {eff.validator_effectiveness = }")

    Args:
        index_or_pubkey: Validator index or pubkey
        from_day: Starting day
        size: Number of results included per page
        follow_next: Whether to follow pagination or not

    Yields:
        Effectiveness metrics
    """
    from_: str | int | date | None = from_day
    if from_ is not None and isinstance(from_, date):
        from_ = from_.isoformat()

    params: Dict[str, Any] = {"from": from_, "size": size}
    url = f"{self.resource_path}/{index_or_pubkey}/effectiveness"
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=ValidatorEffectiveness,
        follow_next=follow_next,
    )

metadata(index_or_pubkey)

Reverse lookup into the entity-to-validator index mappings that live in the RatedDB

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> metadata = eth.validator.metadata(560000)
>>> print(f"{metadata.validator_pubkey = }, {metadata.deposit_addresses = }")

Parameters:

Name Type Description Default
index_or_pubkey int | str

Validator index or pubkey

 required

Returns:

Type Description
ValidatorMetadata

Metadata about the validator
Source code in src/rated/ethereum/validators.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def metadata(self, index_or_pubkey: int | str) -> ValidatorMetadata:
    """
    Reverse lookup into the entity-to-validator index mappings that live in the RatedDB

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> metadata = eth.validator.metadata(560000)
        >>> print(f"{metadata.validator_pubkey = }, {metadata.deposit_addresses = }")

    Args:
        index_or_pubkey: Validator index or pubkey

    Returns:
        Metadata about the validator
    """
    validator = self.client.get(f"{self.resource_path}/{index_or_pubkey}")
    return json_to_instance(validator, ValidatorMetadata)

Validators

Bases: APIResource

Source code in src/rated/ethereum/validators.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class Validators(APIResource):
    path = "/validators"

    def metadata(
        self,
        *,
        from_index: int = 0,
        size: int = 100,
        operators_ids: List[str] | None = None,
        withdrawal_address: str | None = None,
        id_type: IdType = IdType.NODE_OPERATOR,
        follow_next: bool = False,
    ) -> Iterator[ValidatorMetadata]:
        """
        Allows users to request metadata for a group of validators that map to the same operator or pool

        Args:
            from_index: Starting validator index
            size: Number of results included per page
            operators_ids: An array of entities names you want to filter by
            withdrawal_address: Filter by the withdrawal address
            id_type: The type of entity class you would like to filter by
            follow_next: Whether to follow pagination or not

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> metadata = eth.validators.metadata(operators_ids=["Lido", "Kiln"])
            >>> for m in metadata:
            >>>     print(f"{m.validator_pubkey = }, {m.deposit_addresses = }, {m.node_operators = }")

        Yields:
            Validator metadata
        """
        url = f"{self.resource_path}"
        params: Dict[str, Any] = {
            "from": from_index,
            "size": size,
            "operators_ids": operators_ids,
            "withdrawal_address": withdrawal_address,
            "id_type": id_type.value,
        }
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=ValidatorMetadata,
            follow_next=follow_next,
        )

    def effectiveness(
        self,
        *,
        pubkeys: List[str] | None = None,
        indices: List[int] | None = None,
        from_day: int | date | None = None,
        to_day: Union[int, date] | None = None,
        filter_type: FilterType = FilterType.DAY,
        size: int = 10,
        granularity: Granularity | None = None,
        group_by: ValidatorsEffectivenessGroupBy = ValidatorsEffectivenessGroupBy.VALIDATOR,
        follow_next: bool = False,
    ) -> Iterator[ValidatorEffectiveness]:
        """
        Enables the aggregation of all the metrics that live under Validators across an arbitrary number of validator
        indices or pubkeys

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> effectiveness = eth.validators.effectiveness(indices=[500, 501, 502], from_day=795)
            >>> for eff in effectiveness:
            >>>     print(f"{eff.validator_index = }, {eff.validator_effectiveness = }")

        Args:
            pubkeys: Array of pubkeys
            indices: Array of indices
            from_day: Start day
            to_day: End day
            filter_type: Type of filter to apply to from
            size: Number of results included per page
            granularity: The size of time increments you are looking to query
            group_by: Time window or validator; we either group by validator index or across time
            follow_next: Whether to follow pagination or not

        Yields:
            Effectiveness metrics
        """
        from_: str | int | date | None = from_day
        if from_ is not None and isinstance(from_, date):
            from_ = from_.isoformat()

        to_: str | int | date | None = to_day
        if to_ is not None and isinstance(to_, date):
            to_ = to_.isoformat()

        if not pubkeys and not indices:
            raise ValueError("Either pubkeys or indices must be specified")

        if pubkeys and indices:
            raise ValueError("Cannot specify both pubkeys and indices")

        params: Dict[str, Any] = {
            "pubkeys": pubkeys,
            "indices": indices,
            "from": from_,
            "to": to_,
            "filterType": filter_type.value,
            "size": size,
            "granularity": granularity and granularity.value or None,
            "groupBy": group_by.value,
        }
        url = f"{self.resource_path}/effectiveness"
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=ValidatorEffectiveness,
            follow_next=follow_next,
        )

    def report(self, validators: Sequence[str], *, pool_tag: str | None = None) -> int:
        """
        Gateway for node operators to "upload" their sets

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> validator_count = eth.validators.report(["0x...", "0x...", ...], pool_tag="ACME")

        Args:
            validators: Array of validator pubkeys associated with the node operator
            pool_tag: Pool name as they appear in the Rated Explorer

        Returns:
            Number of accepted validators
        """
        url = "/v0/selfReports/validators"
        data = {"validators": validators, "poolTag": pool_tag}
        res = self.client.post(url, json=data)
        count = len(res.json()["validators"])
        return count

effectiveness(*, pubkeys=None, indices=None, from_day=None, to_day=None, filter_type=FilterType.DAY, size=10, granularity=None, group_by=ValidatorsEffectivenessGroupBy.VALIDATOR, follow_next=False)

Enables the aggregation of all the metrics that live under Validators across an arbitrary number of validator indices or pubkeys

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> effectiveness = eth.validators.effectiveness(indices=[500, 501, 502], from_day=795)
>>> for eff in effectiveness:
>>>     print(f"{eff.validator_index = }, {eff.validator_effectiveness = }")

Parameters:

Name Type Description Default
pubkeys List[str] | None

Array of pubkeys

 None
indices List[int] | None

Array of indices

 None
from_day int | date | None

Start day

 None
to_day Union[int, date] | None

End day

 None
filter_type FilterType

Type of filter to apply to from

 DAY
size int

Number of results included per page

 10
granularity Granularity | None

The size of time increments you are looking to query

 None
group_by ValidatorsEffectivenessGroupBy

Time window or validator; we either group by validator index or across time

 VALIDATOR
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
ValidatorEffectiveness

Effectiveness metrics
Source code in src/rated/ethereum/validators.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def effectiveness(
    self,
    *,
    pubkeys: List[str] | None = None,
    indices: List[int] | None = None,
    from_day: int | date | None = None,
    to_day: Union[int, date] | None = None,
    filter_type: FilterType = FilterType.DAY,
    size: int = 10,
    granularity: Granularity | None = None,
    group_by: ValidatorsEffectivenessGroupBy = ValidatorsEffectivenessGroupBy.VALIDATOR,
    follow_next: bool = False,
) -> Iterator[ValidatorEffectiveness]:
    """
    Enables the aggregation of all the metrics that live under Validators across an arbitrary number of validator
    indices or pubkeys

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> effectiveness = eth.validators.effectiveness(indices=[500, 501, 502], from_day=795)
        >>> for eff in effectiveness:
        >>>     print(f"{eff.validator_index = }, {eff.validator_effectiveness = }")

    Args:
        pubkeys: Array of pubkeys
        indices: Array of indices
        from_day: Start day
        to_day: End day
        filter_type: Type of filter to apply to from
        size: Number of results included per page
        granularity: The size of time increments you are looking to query
        group_by: Time window or validator; we either group by validator index or across time
        follow_next: Whether to follow pagination or not

    Yields:
        Effectiveness metrics
    """
    from_: str | int | date | None = from_day
    if from_ is not None and isinstance(from_, date):
        from_ = from_.isoformat()

    to_: str | int | date | None = to_day
    if to_ is not None and isinstance(to_, date):
        to_ = to_.isoformat()

    if not pubkeys and not indices:
        raise ValueError("Either pubkeys or indices must be specified")

    if pubkeys and indices:
        raise ValueError("Cannot specify both pubkeys and indices")

    params: Dict[str, Any] = {
        "pubkeys": pubkeys,
        "indices": indices,
        "from": from_,
        "to": to_,
        "filterType": filter_type.value,
        "size": size,
        "granularity": granularity and granularity.value or None,
        "groupBy": group_by.value,
    }
    url = f"{self.resource_path}/effectiveness"
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=ValidatorEffectiveness,
        follow_next=follow_next,
    )

metadata(*, from_index=0, size=100, operators_ids=None, withdrawal_address=None, id_type=IdType.NODE_OPERATOR, follow_next=False)

Allows users to request metadata for a group of validators that map to the same operator or pool

Parameters:

Name Type Description Default
from_index int

Starting validator index

 0
size int

Number of results included per page

 100
operators_ids List[str] | None

An array of entities names you want to filter by

 None
withdrawal_address str | None

Filter by the withdrawal address

 None
id_type IdType

The type of entity class you would like to filter by

 NODE_OPERATOR
follow_next bool

Whether to follow pagination or not

 False

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> metadata = eth.validators.metadata(operators_ids=["Lido", "Kiln"])
>>> for m in metadata:
>>>     print(f"{m.validator_pubkey = }, {m.deposit_addresses = }, {m.node_operators = }")

Yields:

Type Description
ValidatorMetadata

Validator metadata
Source code in src/rated/ethereum/validators.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def metadata(
    self,
    *,
    from_index: int = 0,
    size: int = 100,
    operators_ids: List[str] | None = None,
    withdrawal_address: str | None = None,
    id_type: IdType = IdType.NODE_OPERATOR,
    follow_next: bool = False,
) -> Iterator[ValidatorMetadata]:
    """
    Allows users to request metadata for a group of validators that map to the same operator or pool

    Args:
        from_index: Starting validator index
        size: Number of results included per page
        operators_ids: An array of entities names you want to filter by
        withdrawal_address: Filter by the withdrawal address
        id_type: The type of entity class you would like to filter by
        follow_next: Whether to follow pagination or not

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> metadata = eth.validators.metadata(operators_ids=["Lido", "Kiln"])
        >>> for m in metadata:
        >>>     print(f"{m.validator_pubkey = }, {m.deposit_addresses = }, {m.node_operators = }")

    Yields:
        Validator metadata
    """
    url = f"{self.resource_path}"
    params: Dict[str, Any] = {
        "from": from_index,
        "size": size,
        "operators_ids": operators_ids,
        "withdrawal_address": withdrawal_address,
        "id_type": id_type.value,
    }
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=ValidatorMetadata,
        follow_next=follow_next,
    )

report(validators, *, pool_tag=None)

Gateway for node operators to "upload" their sets

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> validator_count = eth.validators.report(["0x...", "0x...", ...], pool_tag="ACME")

Parameters:

Name Type Description Default
validators Sequence[str]

Array of validator pubkeys associated with the node operator

 required
pool_tag str | None

Pool name as they appear in the Rated Explorer

 None

Returns:

Type Description
int

Number of accepted validators
Source code in src/rated/ethereum/validators.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def report(self, validators: Sequence[str], *, pool_tag: str | None = None) -> int:
    """
    Gateway for node operators to "upload" their sets

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> validator_count = eth.validators.report(["0x...", "0x...", ...], pool_tag="ACME")

    Args:
        validators: Array of validator pubkeys associated with the node operator
        pool_tag: Pool name as they appear in the Rated Explorer

    Returns:
        Number of accepted validators
    """
    url = "/v0/selfReports/validators"
    data = {"validators": validators, "poolTag": pool_tag}
    res = self.client.post(url, json=data)
    count = len(res.json()["validators"])
    return count

Withdrawals

Bases: APIResource

Offers a view into the future, relating to when a set of withdrawals are expected to land

Source code in src/rated/ethereum/withdrawals.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class Withdrawals(APIResource):
    """Offers a view into the future, relating to when a set of withdrawals are expected to land"""

    path = "/withdrawals"

    def by_operator(
        self,
        operator_id: str,
        *,
        from_day: int | date | None = None,
        size: int | None = None,
        follow_next: bool = False,
    ) -> Iterator[Withdrawal]:
        """
        Retrieve information about the expectation of a withdrawal fulfilment, on a per validator index level

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> withdrawals = eth.withdrawals.by_operator("Lido", from_day=795)
            >>> for w in withdrawals:
            >>>     print(f"{w.withdrawal_slot = }, {w.withdrawable_amount = }")

        Args:
            operator_id: The name of the entity in question
            from_day: Starting day
            size: Number of results included per page
            follow_next: Whether to follow pagination or not

        Yields:
            Withdrawal
        """
        url: str = f"{self.resource_path}/predicted/operators/{operator_id}"
        from_: str | int | date | None = from_day
        if from_ is not None and isinstance(from_, date):
            from_ = from_.isoformat()

        params: Dict[str, Any] = {"from": from_, "size": size}
        return self.client.yield_paginated_results(
            url,
            params=params,
            cls=Withdrawal,
            follow_next=follow_next,
        )

    def by_slot(self, slot: int) -> Iterator[Withdrawal]:
        """
        Returns all the validators that are expected to withdraw by slot

        Examples:
            >>> from rated import Rated
            >>> from rated.ethereum import MAINNET
            >>>
            >>> RATED_KEY = "ey..."
            >>> r = Rated(RATED_KEY)
            >>> eth = r.ethereum(network=MAINNET)
            >>> withdrawals = eth.withdrawals.by_slot(5000)
            >>> for w in withdrawals:
            >>>     print(f"{w.withdrawal_slot = }, {w.id = }, {w.withdrawable_amount = }")

        Args:
            slot: Withdrawal slot number

        Yields:
            Withdrawal
        """
        url: str = f"{self.resource_path}/predicted/slot/{slot}"
        data = self.client.get(url)
        for item in data:
            yield json_to_instance(item, Withdrawal)

by_operator(operator_id, *, from_day=None, size=None, follow_next=False)

Retrieve information about the expectation of a withdrawal fulfilment, on a per validator index level

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> withdrawals = eth.withdrawals.by_operator("Lido", from_day=795)
>>> for w in withdrawals:
>>>     print(f"{w.withdrawal_slot = }, {w.withdrawable_amount = }")

Parameters:

Name Type Description Default
operator_id str

The name of the entity in question

 required
from_day int | date | None

Starting day

 None
size int | None

Number of results included per page

 None
follow_next bool

Whether to follow pagination or not

 False

Yields:

Type Description
Withdrawal

Withdrawal
Source code in src/rated/ethereum/withdrawals.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def by_operator(
    self,
    operator_id: str,
    *,
    from_day: int | date | None = None,
    size: int | None = None,
    follow_next: bool = False,
) -> Iterator[Withdrawal]:
    """
    Retrieve information about the expectation of a withdrawal fulfilment, on a per validator index level

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> withdrawals = eth.withdrawals.by_operator("Lido", from_day=795)
        >>> for w in withdrawals:
        >>>     print(f"{w.withdrawal_slot = }, {w.withdrawable_amount = }")

    Args:
        operator_id: The name of the entity in question
        from_day: Starting day
        size: Number of results included per page
        follow_next: Whether to follow pagination or not

    Yields:
        Withdrawal
    """
    url: str = f"{self.resource_path}/predicted/operators/{operator_id}"
    from_: str | int | date | None = from_day
    if from_ is not None and isinstance(from_, date):
        from_ = from_.isoformat()

    params: Dict[str, Any] = {"from": from_, "size": size}
    return self.client.yield_paginated_results(
        url,
        params=params,
        cls=Withdrawal,
        follow_next=follow_next,
    )

by_slot(slot)

Returns all the validators that are expected to withdraw by slot

Examples:

>>> from rated import Rated
>>> from rated.ethereum import MAINNET
>>>
>>> RATED_KEY = "ey..."
>>> r = Rated(RATED_KEY)
>>> eth = r.ethereum(network=MAINNET)
>>> withdrawals = eth.withdrawals.by_slot(5000)
>>> for w in withdrawals:
>>>     print(f"{w.withdrawal_slot = }, {w.id = }, {w.withdrawable_amount = }")

Parameters:

Name Type Description Default
slot int

Withdrawal slot number

 required

Yields:

Type Description
Withdrawal

Withdrawal
Source code in src/rated/ethereum/withdrawals.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def by_slot(self, slot: int) -> Iterator[Withdrawal]:
    """
    Returns all the validators that are expected to withdraw by slot

    Examples:
        >>> from rated import Rated
        >>> from rated.ethereum import MAINNET
        >>>
        >>> RATED_KEY = "ey..."
        >>> r = Rated(RATED_KEY)
        >>> eth = r.ethereum(network=MAINNET)
        >>> withdrawals = eth.withdrawals.by_slot(5000)
        >>> for w in withdrawals:
        >>>     print(f"{w.withdrawal_slot = }, {w.id = }, {w.withdrawable_amount = }")

    Args:
        slot: Withdrawal slot number

    Yields:
        Withdrawal
    """
    url: str = f"{self.resource_path}/predicted/slot/{slot}"
    data = self.client.get(url)
    for item in data:
        yield json_to_instance(item, Withdrawal)