File size: 7,115 Bytes
cbb819c |
1 2 3 4 5 6 7 8 9 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 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 |
import json
import requests
from web3 import Web3
from phi.tools import Toolkit
from phi.utils.log import logger
from src.libs.web3 import get_web3_instance
from src.libs.helper_functions import get_headers, get_private_key
from src.libs.token_approval_helper import TokenApprovalHelper
class CryptoSwapTools(Toolkit):
def __init__(self, web3: Web3 = get_web3_instance()):
super().__init__(name="swap_tools")
# Store Web3 instance
self.web3 = web3
# Helper for token approval
self.token_approval_helper = TokenApprovalHelper(web3, get_private_key())
# Registering methods to make them accessible via the toolkit
self.register(self.get_swap_quote)
self.register(self.get_swap_price)
self.register(self.get_swap_sources)
self.register(self.execute_swap)
def get_swap_quote(self, buy_token: str, sell_token: str, sell_amount: str) -> str:
"""
Fetches a swap quote from the 0x Swap API.
Args:
buy_token (str): The token to buy (e.g., 'DAI').
sell_token (str): The token to sell (e.g., 'ETH').
sell_amount (str): The amount of the sell token to swap, in the smallest unit (e.g., wei for ETH).
Returns:
dict: A dictionary containing the swap quote details.
Example:
>>> get_swap_quote('DAI', 'ETH', '1000000000000000000')
"""
logger.info(f"Fetching swap quote: buying {buy_token} with {sell_token} amount {sell_amount}")
try:
params = {
'buyToken': buy_token,
'sellToken': sell_token,
'sellAmount': sell_amount
}
response = requests.get("https://api.0x.org/swap/v1/quote", params=params, headers=get_headers())
response.raise_for_status()
# return response.json()
return f"{(response.json())}"
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to get swap quote: {e}")
# return {"error": str(e)}
return f"Error: {e}"
def get_swap_price(self, buy_token: str, sell_token: str, buy_amount: str) -> str:
"""
Fetches the price for a swap from the 0x Swap API.
Args:
buy_token (str): The token to buy.
sell_token (str): The token to sell.
buy_amount (str): The amount of the buy token, in the smallest unit.
Returns:
dict: A dictionary containing the swap price details.
Example:
>>> get_swap_price('DAI', 'ETH', '1000000000000000000')
"""
logger.info(f"Fetching swap price: buying {buy_token} with {sell_token} amount {buy_amount}")
try:
params = {
'buyToken': buy_token,
'sellToken': sell_token,
'buyAmount': buy_amount
}
response = requests.get("https://api.0x.org/swap/v1/price", params=params, headers=get_headers())
response.raise_for_status()
# return response.json()
return f"{(response.json())}"
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to get swap price: {e}")
# return {"error": str(e)}
return f"Error: {e}"
def get_swap_sources(self) -> str:
"""
Fetches the list of liquidity sources from the 0x Swap API.
Returns:
dict: A dictionary containing the list of liquidity sources.
Example:
>>> get_swap_sources()
"""
logger.info("Fetching swap sources")
try:
response = requests.get("https://api.0x.org/swap/v1/sources", headers=get_headers())
response.raise_for_status()
# return response.json()
return f"{(response.json())}"
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to get swap sources: {e}")
# return {"error": str(e)}
return f"Error: {e}"
def execute_swap(self, buy_token: str, sell_token: str, sell_amount: str, eth_address: str) -> str:
"""
Executes a swap using the 0x Swap API.
Args:
buy_token (str): The token to buy (e.g., 'DAI').
sell_token (str): The token to sell (e.g., 'ETH').
sell_amount (str): The amount of the sell token to swap, in the smallest unit (e.g., wei for ETH).
eth_address (str): The Ethereum address of the user executing the swap.
Returns:
dict: The transaction receipt of the swap transaction.
Example:
>>> execute_swap('DAI', 'ETH', '1000000000000000000', '0xYourEthereumAddress')
"""
# Get the swap quote
quote = json.loads(self.get_swap_quote(buy_token, sell_token, sell_amount))
logger.info(f"Swap quote: {quote}")
if 'error' in quote:
# return {"error": "Failed to get swap quote"}
return f"Error: Failed to get swap quote"
# Approve the token if needed (skip if selling ETH)
if sell_token != 'ETH':
approval_receipt = self.token_approval_helper.approve_token(sell_token, quote['allowanceTarget'],
sell_amount, eth_address)
logger.info(f"Approval receipt: {approval_receipt}")
if 'status' not in approval_receipt or approval_receipt['status'] != 1:
# return {"error": "Token approval failed"}
return f"Error: Token approval failed"
# Execute the swap
try:
swap_tx = {
'from': eth_address,
'to': quote['to'],
'data': quote['data'],
'value': int(quote['value']),
'gas': 200000,
'gasPrice': self.web3.to_wei('20', 'gwei'),
'nonce': self.web3.eth.get_transaction_count(eth_address)
}
signed_swap_tx = self.web3.eth.account.signTransaction(swap_tx, private_key=get_private_key())
tx_hash = self.web3.eth.send_raw_transaction(signed_swap_tx.rawTransaction)
receipt = self.web3.eth.wait_for_transaction_receipt(tx_hash)
logger.info(f"Swap transaction receipt: {receipt}")
# return receipt
return f"{(receipt)}"
except Exception as e:
logger.warning(f"Failed to execute swap: {e}")
# return {"error": str(e)}
return f"Error: {e}"
# Example usage
if __name__ == "__main__":
eth_address = '0xYourEthereumAddress' # Replace with your Ethereum address
# Initialize Web3
web3 = Web3(Web3.HTTPProvider('YOUR_INFURA_OR_ALCHEMY_URL'))
# Initialize CryptoSwapTools
swap_tool = CryptoSwapTools(web3)
# Fetch and print swap sources
sources = swap_tool.get_swap_sources()
print(f"Swap Sources: {sources}")
# Execute a swap (example)
receipt = swap_tool.execute_swap('DAI', 'ETH', '1000000000000000000', eth_address)
print(f"Swap Transaction Receipt: {receipt}")
|