convert_tokens
Function
def convert_tokens(account, token0_address, token1_address, output_amount, attempts=18):
# check if conversion route exists
routes_functions = json.load(open('./data/routes.json'))
if token0_address not in routes_functions[token1_address]['functions'].keys():
raise Exception("Route not available for {} to {}".format(token0_address, token1_address))
# get the cost required to convert tokens
cost = routes_functions[token1_address]['costs'][token0_address]
tokens_required = cost * output_amount
if (tokens_balance := get_token_balance(token0_address, account.address)) < tokens_required:
logging.error("Need {} more tokens".format(tokens_required - tokens_balance))
return False
# call the buy function with amount or default to no args
call_function = routes_functions[token1_address]['functions'][token0_address]
approve_token_spending(account, token0_address, token1_address, get_token_supply(token0_address, True))
token1_contract = load_contract(token1_address, load_contract_abi(token1_address))
amount = to_token_decimals(output_amount, token1_contract.functions.decimals().call())
try:
tx = getattr(token1_contract.functions, call_function)(int(amount)).build_transaction({
"from": account.address,
"nonce": get_nonce(account.address)
})
except Web3ValidationError as e:
if 'positional arguments with type(s) `int`' in str(e):
for i in range(0, amount):
# cancel the rest of this loop if the gas price is too damn high
if get_beacon_gas_prices('rapid', beacon_gasnow_cache_seconds) > rapid_gas_fee_limit:
logging.warning("Gas fees are too high")
return None
try:
tx = getattr(token1_contract.functions, call_function)().build_transaction({
"from": account.address,
"nonce": get_nonce(account.address)
})
success = broadcast_transaction(account, tx, True, attempts)
except Exception as e:
if error := interpret_exception_message(e):
logging.error(
"{}. Failed to convert using {}".format(error, routes_functions[token1_address]['label']))
else:
if success:
logging.info("Called {}({}) from {}".format(
call_function,
amount,
routes_functions[token1_address]['label']
))
else:
logging.warning("Failed to call {}({}) from {}".format(
call_function,
amount,
routes_functions[token1_address]['label']
))
return False
else:
raise Web3ValidationError(e)
return True
else:
try:
success = broadcast_transaction(account, tx, True, attempts)
except Exception as e:
if error := interpret_exception_message(e):
logging.error("{}. Failed to convert using {}".format(error, routes_functions[token1_address]['label']))
return False
else:
if success:
logging.info("Called {}({}) from {}".format(
call_function,
amount,
routes_functions[token1_address]['label']
))
return True
else:
logging.warning("Failed to call {}({}) from {}".format(
call_function,
amount,
routes_functions[token1_address]['label']
))
return False