tradehub.tradescan_client.TradescanClient.get_liquidations() - Code Metrics - Inspection of "Feature/tradscan api" - Mai-Te-Pora/tradehub-python - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#28)

by Switcheolytics

created 2021-02-24 21:23 UTC

TradescanClient.get_liquidations() A

↳ Parent: tradehub.tradescan_client

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
rs	10
c	0
b	0
f	0

"""
Description:
    Public Client for the TradeScan API content you see on Switcheo.org
    Its not required to use your private key to interact with these endpoints to track the blockchain state.
Usage:
    from tradescan.public_client import PublicClient
"""

from tradehub.decentralized_client import NetworkCrawlerClient


class TradescanClient(NetworkCrawlerClient):
    """
    This class allows the user to interact with the TradeScan API including information
    available with validators, tokens, delegators, addresses, and blockchain stats.
    """

    def __init__(self,
                 network: str = 'testnet',
                 trusted_ips: list = None,
                 trusted_uris: list = None):
        """
        :param api_url: The URL for the Switcheo API endpoint.
        :type api_url: str
        """
        if trusted_ips and trusted_uris:
            raise ValueError("Use IP's or URI's, not both!")

        NetworkCrawlerClient.__init__(self, network=network, trusted_ip_list=trusted_ips, trusted_uri_list=trusted_uris)

    def get_address_rewards(self, address):
        if address is not None and isinstance(address, str):
            return self.tradehub_get_request(path='/distribution/delegators/{}/rewards'.format(address))

    def get_address_staking(self, address):
        if address is not None and isinstance(address, str):
            return self.tradehub_get_request(path='/staking/delegators/{}/delegations'.format(address))

    def get_address_trades(self, limit=200, pagination=None, address=None):
        api_params = {}
        if limit is not None and limit <= 1000:
            api_params["limit"] = limit
        if pagination is not None and pagination in [True, False]:
            api_params["pagination"] = pagination
        if address is not None and isinstance(address, str):
            api_params["account"] = address
        return self.tradehub_get_request(path='/get_trades_by_account', params=api_params)

    def get_all_validators(self):
        return self.tradehub_get_request(path='/get_all_validators')

    def get_balance(self, address):
        api_params = {}
        if address is not None and isinstance(address, str):
            api_params["account"] = address
        return self.tradehub_get_request(path='/get_balance', params=api_params)

    def get_block(self, block_nbr=0):
        if block_nbr is not None and isinstance(block_nbr, int) and block_nbr > 0:
            return self.tradehub_get_request(path='/blocks/{}'.format(block_nbr))

    def get_blocks(self, limit=200, pagination=None, validator_address=None):
        api_params = {}
        if limit is not None and limit <= 1000:
            api_params["limit"] = limit
        if pagination is not None and pagination in [True, False]:
            api_params["pagination"] = pagination
        if validator_address is not None:
            api_params["proposer"] = validator_address
        return self.tradehub_get_request(path='/get_blocks', params=api_params)

    def get_block_time(self):
        return self.tradehub_get_request(path='/get_block_time')

    def get_commitment_curve(self):
        return self.tradehub_get_request(path='/get_commitment_curve')

    def get_distribution_parameters(self):
        return self.tradehub_get_request(path='/distribution/parameters')

    def get_external_transfers(self, address):
        api_params = {}
        if address is not None and isinstance(address, str):
            api_params["account"] = address
        return self.tradehub_get_request(path='/get_external_transfers', params=api_params)

    def get_inflation_start_time(self):
        return self.tradehub_get_request(path='/get_inflation_start_time')

    def get_latest_blocks(self):
        return self.tradehub_get_request(path='/blocks/latest')

    def get_liquidity_pools(self):
        return self.tradehub_get_request(path='/get_liquidity_pools')

    def get_liquidations(self):
        return self.tradehub_get_request(path='/get_liquidations')

    def get_markets(self,
                    limit=200,
                    pagination=None,
                    is_settled=None):
        api_params = {}
        if limit is not None and limit <= 1000:
            api_params["limit"] = limit
        if pagination is not None and pagination in [True, False]:
            api_params["pagination"] = pagination
        if is_settled is not None and is_settled in [True, False]:
            api_params["is_settled"] = is_settled
        return self.tradehub_get_request(path='/get_markets', params=api_params)

    def get_orders(self,
                   limit=200,
                   pagination=None,
                   transaction_type=None,
                   block_nbr=None,
                   initiator=None,
                   order_status=None,
                   order_type=None):
        api_params = {}
        if limit is not None and limit <= 1000:
            api_params["limit"] = limit
        if pagination is not None and pagination in [True, False]:
            api_params["pagination"] = pagination
        if transaction_type is not None and transaction_type in self.transaction_types:
            api_params["msg_type"] = transaction_type
        if block_nbr is not None and isinstance(block_nbr, int) and block_nbr > 0:
            api_params["height"] = block_nbr
        if initiator is not None:              # system_derisk, system_adl, system_liquidate
            api_params["initiator"] = initiator
        if order_status is not None:           # open
            api_params["order_status"] = order_status
        if order_type is not None:             # market, liquidation
            api_params["order_type"] = order_type
        return self.tradehub_get_request(path='/get_orders', params=api_params)

    def get_positions(self, address):
        api_params = {}
        if address is not None and isinstance(address, str):
            api_params["account"] = address
        return self.tradehub_get_request(path='/get_positions', params=api_params)

    def get_profile(self, address):
        api_params = {}
        if address is not None and isinstance(address, str):
            api_params["account"] = address
        return self.tradehub_get_request(path='/get_profile', params=api_params)

    def get_reward_curve(self):
        return self.tradehub_get_request(path='/get_reward_curve')

    def get_rich_list(self, token):
        api_params = {}
        if token is not None and token.lower() in self.tokens:
            api_params["token"] = token.lower()
        return self.tradehub_get_request(path='/get_rich_list', params=api_params)

    def get_staking_pool(self):
        return self.tradehub_get_request(path='/staking/pool')

    def get_token(self, token):
        api_params = {}
        if token is not None and token.lower() in self.tokens:
            api_params["token"] = token.lower()
        return self.tradehub_get_request(path='/token', params=api_params)

    def get_token_list(self):
        token_list = []
        tokens = self.get_tokens()
        for token in tokens:
            token_list.append(token["denom"])
        return token_list

    def get_tokens(self):
        return self.tradehub_get_request(path='/get_tokens')

    def get_total_balances(self):
        return self.tradehub_get_request(path='/get_total_balances')

    def get_tradehub_monitor(self):
        return self.tradehub_get_request(path='/monitor')

    def get_transaction(self, transaction_hash=None):
        api_params = {}
        if transaction_hash is not None and isinstance(transaction_hash, str):
            api_params["hash"] = transaction_hash
        return self.tradehub_get_request(path='/get_transaction', params=api_params)

    def get_transaction_fees(self):
        return self.tradehub_get_request(path='/get_txns_fees')

    def get_transaction_log(self, transaction_hash=None):
        api_params = {}
        if transaction_hash is not None and isinstance(transaction_hash, str):
            api_params["hash"] = transaction_hash
        return self.tradehub_get_request(path='/get_tx_log', params=api_params)

    def get_transaction_types(self):
        return self.tradehub_get_request(path='/get_transaction_types')

    def get_transactions(self,
                         limit=200,
                         pagination=None,
                         address=None,
                         transaction_type=None,
                         block_nbr=None):
        api_params = {}
        if limit is not None and limit <= 1000:
            api_params["limit"] = limit
        if pagination is not None and pagination in [True, False]:
            api_params["pagination"] = pagination
        if address is not None and isinstance(address, str):
            api_params["address"] = address
        if transaction_type is not None and (transaction_type in self.transaction_types or transaction_type == ''):
            api_params["msg_type"] = transaction_type
        if block_nbr is not None and isinstance(block_nbr, int) and block_nbr > 0:
            api_params["height"] = block_nbr
        return self.tradehub_get_request(path='/get_transactions', params=api_params)

    def get_validator_delegations(self, operator_address):
        return self.tradehub_get_request(path='/staking/validators/{}/delegations'.format(operator_address))

    def get_validator_public_nodes(self):
        public_nodes = {}
        tradehub_state = self.get_tradehub_monitor()
        for validator in tradehub_state:
            public_nodes[validator["moniker"]] = validator["ip"]
        return public_nodes

    def get_validator_public_node_ips(self):
        public_node_ips = []
        nodes_dict = self.get_validator_public_nodes()
        for key in nodes_dict.keys():
            public_node_ips.append(nodes_dict[key])
        return public_node_ips

    def get_validator_missed_blocks(self, address=None):
        validators_missed_blocks = {}
        validators_signing_info = self.get_validator_signing_info()["result"]
        for validator_signing_info in validators_signing_info:
            validators_missed_blocks[validator_signing_info["address"]] = validator_signing_info["missed_blocks_counter"]
        print(validators_missed_blocks)
        return validators_missed_blocks[address]

    def get_validator_signing_info(self, limit=100):
        api_params = {}
        if limit is not None and isinstance(limit, int) and limit > 0:
            api_params["limit"] = limit
        return self.tradehub_get_request(path='/slashing/signing_infos', params=api_params)

    def get_validators(self, status=None):
        api_params = {}
        if status is not None and status in ["unbonding", "unbonded"]:
            api_params["status"] = status
        return self.tradehub_get_request(path='/staking/validators', params=api_params)


1			"""
2			Description:
3			Public Client for the TradeScan API content you see on Switcheo.org
4			Its not required to use your private key to interact with these endpoints to track the blockchain state.
5			Usage:
6			from tradescan.public_client import PublicClient
7			"""
8
9			from tradehub.decentralized_client import NetworkCrawlerClient
10
11
12			class TradescanClient(NetworkCrawlerClient):
13			"""
14			This class allows the user to interact with the TradeScan API including information
15			available with validators, tokens, delegators, addresses, and blockchain stats.
16			"""
17
18			def __init__(self,
19			network: str = 'testnet',
20			trusted_ips: list = None,
21			trusted_uris: list = None):
22			"""
23			:param api_url: The URL for the Switcheo API endpoint.
24			:type api_url: str
25			"""
26			if trusted_ips and trusted_uris:
27			raise ValueError("Use IP's or URI's, not both!")
28
29			NetworkCrawlerClient.__init__(self, network=network, trusted_ip_list=trusted_ips, trusted_uri_list=trusted_uris)
30
31			def get_address_rewards(self, address):
32			if address is not None and isinstance(address, str):
33			return self.tradehub_get_request(path='/distribution/delegators/{}/rewards'.format(address))
34
35			def get_address_staking(self, address):
36			if address is not None and isinstance(address, str):
37			return self.tradehub_get_request(path='/staking/delegators/{}/delegations'.format(address))
38
39			def get_address_trades(self, limit=200, pagination=None, address=None):
40			api_params = {}
41			if limit is not None and limit <= 1000:
42			api_params["limit"] = limit
43			if pagination is not None and pagination in [True, False]:
44			api_params["pagination"] = pagination
45			if address is not None and isinstance(address, str):
46			api_params["account"] = address
47			return self.tradehub_get_request(path='/get_trades_by_account', params=api_params)
48
49			def get_all_validators(self):
50			return self.tradehub_get_request(path='/get_all_validators')
51
52			def get_balance(self, address):
53			api_params = {}
54			if address is not None and isinstance(address, str):
55			api_params["account"] = address
56			return self.tradehub_get_request(path='/get_balance', params=api_params)
57
58			def get_block(self, block_nbr=0):
59			if block_nbr is not None and isinstance(block_nbr, int) and block_nbr > 0:
60			return self.tradehub_get_request(path='/blocks/{}'.format(block_nbr))
61
62			def get_blocks(self, limit=200, pagination=None, validator_address=None):
63			api_params = {}
64			if limit is not None and limit <= 1000:
65			api_params["limit"] = limit
66			if pagination is not None and pagination in [True, False]:
67			api_params["pagination"] = pagination
68			if validator_address is not None:
69			api_params["proposer"] = validator_address
70			return self.tradehub_get_request(path='/get_blocks', params=api_params)
71
72			def get_block_time(self):
73			return self.tradehub_get_request(path='/get_block_time')
74
75			def get_commitment_curve(self):
76			return self.tradehub_get_request(path='/get_commitment_curve')
77
78			def get_distribution_parameters(self):
79			return self.tradehub_get_request(path='/distribution/parameters')
80
81			def get_external_transfers(self, address):
82			api_params = {}
83			if address is not None and isinstance(address, str):
84			api_params["account"] = address
85			return self.tradehub_get_request(path='/get_external_transfers', params=api_params)
86
87			def get_inflation_start_time(self):
88			return self.tradehub_get_request(path='/get_inflation_start_time')
89
90			def get_latest_blocks(self):
91			return self.tradehub_get_request(path='/blocks/latest')
92
93			def get_liquidity_pools(self):
94			return self.tradehub_get_request(path='/get_liquidity_pools')
95
96			def get_liquidations(self):
97			return self.tradehub_get_request(path='/get_liquidations')
98
99			def get_markets(self,
100			limit=200,
101			pagination=None,
102			is_settled=None):
103			api_params = {}
104			if limit is not None and limit <= 1000:
105			api_params["limit"] = limit
106			if pagination is not None and pagination in [True, False]:
107			api_params["pagination"] = pagination
108			if is_settled is not None and is_settled in [True, False]:
109			api_params["is_settled"] = is_settled
110			return self.tradehub_get_request(path='/get_markets', params=api_params)
111
112			def get_orders(self,
113			limit=200,
114			pagination=None,
115			transaction_type=None,
116			block_nbr=None,
117			initiator=None,
118			order_status=None,
119			order_type=None):
120			api_params = {}
121			if limit is not None and limit <= 1000:
122			api_params["limit"] = limit
123			if pagination is not None and pagination in [True, False]:
124			api_params["pagination"] = pagination
125			if transaction_type is not None and transaction_type in self.transaction_types:
126			api_params["msg_type"] = transaction_type
127			if block_nbr is not None and isinstance(block_nbr, int) and block_nbr > 0:
128			api_params["height"] = block_nbr
129			if initiator is not None: # system_derisk, system_adl, system_liquidate
130			api_params["initiator"] = initiator
131			if order_status is not None: # open
132			api_params["order_status"] = order_status
133			if order_type is not None: # market, liquidation
134			api_params["order_type"] = order_type
135			return self.tradehub_get_request(path='/get_orders', params=api_params)
136
137			def get_positions(self, address):
138			api_params = {}
139			if address is not None and isinstance(address, str):
140			api_params["account"] = address
141			return self.tradehub_get_request(path='/get_positions', params=api_params)
142
143			def get_profile(self, address):
144			api_params = {}
145			if address is not None and isinstance(address, str):
146			api_params["account"] = address
147			return self.tradehub_get_request(path='/get_profile', params=api_params)
148
149			def get_reward_curve(self):
150			return self.tradehub_get_request(path='/get_reward_curve')
151
152			def get_rich_list(self, token):
153			api_params = {}
154			if token is not None and token.lower() in self.tokens:
155			api_params["token"] = token.lower()
156			return self.tradehub_get_request(path='/get_rich_list', params=api_params)
157
158			def get_staking_pool(self):
159			return self.tradehub_get_request(path='/staking/pool')
160
161			def get_token(self, token):
162			api_params = {}
163			if token is not None and token.lower() in self.tokens:
164			api_params["token"] = token.lower()
165			return self.tradehub_get_request(path='/token', params=api_params)
166
167			def get_token_list(self):
168			token_list = []
169			tokens = self.get_tokens()
170			for token in tokens:
171			token_list.append(token["denom"])
172			return token_list
173
174			def get_tokens(self):
175			return self.tradehub_get_request(path='/get_tokens')
176
177			def get_total_balances(self):
178			return self.tradehub_get_request(path='/get_total_balances')
179
180			def get_tradehub_monitor(self):
181			return self.tradehub_get_request(path='/monitor')
182
183			def get_transaction(self, transaction_hash=None):
184			api_params = {}
185			if transaction_hash is not None and isinstance(transaction_hash, str):
186			api_params["hash"] = transaction_hash
187			return self.tradehub_get_request(path='/get_transaction', params=api_params)
188
189			def get_transaction_fees(self):
190			return self.tradehub_get_request(path='/get_txns_fees')
191
192			def get_transaction_log(self, transaction_hash=None):
193			api_params = {}
194			if transaction_hash is not None and isinstance(transaction_hash, str):
195			api_params["hash"] = transaction_hash
196			return self.tradehub_get_request(path='/get_tx_log', params=api_params)
197
198			def get_transaction_types(self):
199			return self.tradehub_get_request(path='/get_transaction_types')
200
201			def get_transactions(self,
202			limit=200,
203			pagination=None,
204			address=None,
205			transaction_type=None,
206			block_nbr=None):
207			api_params = {}
208			if limit is not None and limit <= 1000:
209			api_params["limit"] = limit
210			if pagination is not None and pagination in [True, False]:
211			api_params["pagination"] = pagination
212			if address is not None and isinstance(address, str):
213			api_params["address"] = address
214			if transaction_type is not None and (transaction_type in self.transaction_types or transaction_type == ''):
215			api_params["msg_type"] = transaction_type
216			if block_nbr is not None and isinstance(block_nbr, int) and block_nbr > 0:
217			api_params["height"] = block_nbr
218			return self.tradehub_get_request(path='/get_transactions', params=api_params)
219
220			def get_validator_delegations(self, operator_address):
221			return self.tradehub_get_request(path='/staking/validators/{}/delegations'.format(operator_address))
222
223			def get_validator_public_nodes(self):
224			public_nodes = {}
225			tradehub_state = self.get_tradehub_monitor()
226			for validator in tradehub_state:
227			public_nodes[validator["moniker"]] = validator["ip"]
228			return public_nodes
229
230			def get_validator_public_node_ips(self):
231			public_node_ips = []
232			nodes_dict = self.get_validator_public_nodes()
233			for key in nodes_dict.keys():
234			public_node_ips.append(nodes_dict[key])
235			return public_node_ips
236
237			def get_validator_missed_blocks(self, address=None):
238			validators_missed_blocks = {}
239			validators_signing_info = self.get_validator_signing_info()["result"]
240			for validator_signing_info in validators_signing_info:
241			validators_missed_blocks[validator_signing_info["address"]] = validator_signing_info["missed_blocks_counter"]
242			print(validators_missed_blocks)
243			return validators_missed_blocks[address]
244
245			def get_validator_signing_info(self, limit=100):
246			api_params = {}
247			if limit is not None and isinstance(limit, int) and limit > 0:
248			api_params["limit"] = limit
249			return self.tradehub_get_request(path='/slashing/signing_infos', params=api_params)
250
251			def get_validators(self, status=None):
252			api_params = {}
253			if status is not None and status in ["unbonding", "unbonded"]:
254			api_params["status"] = status
255			return self.tradehub_get_request(path='/staking/validators', params=api_params)
256

Mai-Te-Pora / tradehub-python

Pull Request — main (#28)

TradescanClient.get_liquidations() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like