Fetching Gas Prices Script
This script handles the process of fetching gas prices for hedger transactions from different sources, checks for errors, and ensures the gas price does not exceed a specified upper bound.
Full Script:
def __get_gas_from_metaswap(self, priority, gas_upper_bound):
gas_provider = f'https://gas-api.metaswap.codefi.network/networks/{self.chain_id}/suggestedGasFees'
resp = None
try:
resp = requests.get(gas_provider, timeout=RequestTimeout)
resp_json = resp.json()
max_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxFeePerGas'])
max_priority_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxPriorityFeePerGas'])
apm.span_label(max_fee_per_gas=max_fee_per_gas, max_priority_fee_per_gas=max_priority_fee_per_gas,
gas_price_provider=gas_provider)
if max_fee_per_gas > gas_upper_bound:
raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {max_fee_per_gas}')
gas_params = {
'maxFeePerGas': Web3.to_wei(max_fee_per_gas, 'GWei'),
'maxPriorityFeePerGas': Web3.to_wei(max_priority_fee_per_gas, 'GWei')
}
return gas_params
except (RequestException, JSONDecodeError, KeyError):
if not DevEnv:
logging.exception(f'Failed to get gas info from metaswap {resp.status_code=}')
raise FailedToGetGasPrice("Failed to get gas info from metaswap")
# fixme: do it by asyncio way
def __get_gas_from_rpc(self, priority, gas_upper_bound):
gas_price = None
for provider in self.providers:
provider_url = str(provider.provider.endpoint_uri) # noqa
try:
gas_price = provider.eth.gas_price
apm.span_label(gas_price_from_provider=str(gas_price / 1e9), gas_price_provider=provider_url)
except (ConnectionError, ReadTimeout, ValueError) as e:
logging.error(f'Failed to get gas price from {provider_url}, {e=}')
continue
break
if gas_price is None:
raise FailedToGetGasPrice("Non of RCP could provide gas price!")
if gas_price / 1e9 > gas_upper_bound:
raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {gas_price / 1e9}')
return dict(gasPrice=int(gas_price * TxPriorityMultiplier.get(priority, 1)))
def _get_gas_price(self, gas_upper_bound, priority) -> dict:
gas_params = {}
if self.chain_id == 97: # Test BNB Network
gas_params['gasPrice'] = Web3.to_wei(10.1, 'GWei')
elif DevEnv or self.chain_id in ChainIdsWithGasFromRPC:
gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)
else:
try:
gas_params = self.__get_gas_from_metaswap(priority, gas_upper_bound)
except FailedToGetGasPrice:
gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)
apm.span_label(**gas_params)
return gas_params
__get_gas_from_metaswap
Function Definition and Parameters:
def __get_gas_from_metaswap(self, priority, gas_upper_bound):
This function is designed to retrieve gas price information from the Metamask Gas API. It takes two parameters:
priority
: The priority level of the transaction, which will determine the gas fee structure to use (low, medium, high).gas_upper_bound
: The maximum allowable gas price for the transaction.
Gas Provider URL:
gas_provider = f'https://gas-api.metaswap.codefi.network/networks/{self.chain_id}/suggestedGasFees'
Constructs the URL to fetch gas price information from the Metamask Gas API for the specific blockchain network identified by self.chain_id
.
Initialize Response:
Initializes the variable resp
to None
. This will later be used to store the response from the API request.
Fetching and Processing Data:
try:
resp = requests.get(gas_provider, timeout=RequestTimeout)
resp_json = resp.json()
Attempts to send a GET request to the gas_provider
URL with a specified timeout (RequestTimeout
). The response from this request is stored in resp
,then converts the response from the API into JSON format, which allows for easy access to the data fields within the response.
Extract Gas Fees:
max_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxFeePerGas'])
max_priority_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxPriorityFeePerGas'])
Extracts the suggested maximum fee per gas unit (suggestedMaxFeePerGas) and the suggested maximum priority fee per gas unit (Decimal) from the JSON response, converting these values to Decimal.
Log Span Label:
apm.span_label(max_fee_per_gas=max_fee_per_gas, max_priority_fee_per_gas=max_priority_fee_per_gas, gas_price_provider=gas_provider)
Logs the extracted gas fee information using apm.span_label for monitoring purposes. This includes the maximum fee per gas, the maximum priority fee per gas, and the URL of the gas price provider.
Check Upper Bound:
if max_fee_per_gas > gas_upper_bound:
raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {max_fee_per_gas}')
Checks if the max_fee_per_gas
exceeds the specified gas_upper_bound
. If it does, raises an OutOfRangeTransactionFee
exception with a descriptive message.
Prepare Gas Parameters:
gas_params = {
'maxFeePerGas': Web3.to_wei(max_fee_per_gas, 'GWei'),
'maxPriorityFeePerGas': Web3.to_wei(max_priority_fee_per_gas, 'GWei')
}
return gas_params
If the gas fee is within the acceptable range, this prepares a dictionary gas_params containing the gas fee values converted to Wei (a smaller unit of Ether) using Web3.to_wei, then returns the gas_params dictionary.
Exception Handling:
except (RequestException, JSONDecodeError, KeyError):
if not DevEnv:
logging.exception(f'Failed to get gas info from metaswap {resp.status_code=}')
raise FailedToGetGasPrice("Failed to get gas info from metaswap")
Handles potential exceptions that might occur during the process:
RequestException
: General exception for request issues.JSONDecodeError
: Issue decoding the JSON response.KeyError
: If the expected keys are not present in the JSON response.
If an exception occurs, logs the error (unless in development environment DevEnv) and raises a FailedToGetGasPrice exception with an appropriate message.
__get_gas_from_rpc
Function Definition and Parameters:
def __get_gas_from_rpc(self, priority, gas_upper_bound):
This function is a private method to get gas prices from an RPC provider. It takes priority and gas_upper_bound as parameters.
Initialize Gas Price:
gas_price = None
Initializes the gas price variable to None.
Loop Through Providers:
for provider in self.providers:
provider_url = str(provider.provider.endpoint_uri) # noqa
try:
gas_price = provider.eth.gas_price
apm.span_label(gas_price_from_provider=str(gas_price / 1e9), gas_price_provider=provider_url)
except (ConnectionError, ReadTimeout, ValueError) as e:
logging.error(f'Failed to get gas price from {provider_url}, {e=}')
continue
break
Iterates through available RPC providers to fetch the gas price. If it fails for a provider, logs the error and continues to the next one. Also logs the gas fees and provider URL for tracing purposes with apm.span_label
Check Gas Price:
if gas_price is None:
raise FailedToGetGasPrice("None of RCP could provide gas price!")
if gas_price / 1e9 > gas_upper_bound:
raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {gas_price / 1e9}')
If a gas price was retrieved, the function checks whether this price divided by 1e9 (to convert from wei to GWei) exceeds the gas_upper_bound
. If it does, it raises an exception for exceeding the specified gas price limit.
Return Gas Parameters:
return dict(gasPrice=int(gas_price * TxPriorityMultiplier.get(priority, 1)))
Finally, the function returns a dictionary with a single key-value pair. The key is gasPrice, and the value is the fetched gas_price adjusted by a priority multiplier. This multiplier adjusts the gas price based on the transaction priority, allowing for some dynamic control over how much to pay per transaction.
__get_gas_price
Function Definition and Parameters:
def _get_gas_price(self, gas_upper_bound, priority) -> dict:
This function fetches the gas price based on the specified gas_upper_bound and priority.
Initialize Gas Parameters:
gas_params = {}
Initializes an empty dictionary gas_params
to store the gas price parameters.
Conditional Check for Test BNB Network
if self.chain_id == 97: # Test BNB Network
gas_params['gasPrice'] = Web3.to_wei(10.1, 'GWei')
Checks if the chain_id
is 97, which corresponds to the Test BNB Network. If true, it sets the gasPrice to 10.1 GWei (converted to Wei) directly in the gas_params dictionary.
Conditional Check for Development Environment or Specific Chains
elif DevEnv or self.chain_id in ChainIdsWithGasFromRPC:
gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)
If the code is running in a development environment (DevEnv) or if the chain_id
is in the ChainIdsWithGasFromRPC
list, it fetches the gas price from RPC providers by calling the __get_gas_from_rpc
method with the specified priority and gas_upper_bound
.
Attempt to Get Gas Price from Metamask Swap API
else:
try:
gas_params = self.__get_gas_from_metaswap(priority, gas_upper_bound)
except FailedToGetGasPrice:
gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)
If the previous conditions are not met, it attempts to fetch the gas price from MetaSwap using the __get_gas_from_metaswap
method. If this attempt fails and raises a FailedToGetGasPrice
exception, it falls back to fetching the gas price from RPC providers using the __get_gas_from_rpc
method.
Logging the Gas Parameters
apm.span_label(**gas_params)
Logs the gas price parameters using for monitoring purposes. The **gas_params syntax unpacks the dictionary so that each key-value pair in gas_params is passed as a separate keyword argument.
Returning the Gas Parameters
apm.span_label(**gas_params)
Returns the gas_params dictionary to the caller.
Summary:
The _get_gas_price
function is designed to determine and return the appropriate gas price parameters for a transaction. It first checks if the network is the Test BNB Network, in which case it directly sets the gas price. If the environment is a development environment or the chain ID is in a specific list, it fetches the gas price from RPC providers. Otherwise, it tries to fetch the gas price from MetaSwap and falls back to RPC providers if that fails.
Last updated