add uniswap v3 arbitrage

external_workers/common/utils.py

@@ -0,0 +1,17 @@
+import uuid
+import os
+from typing import Callable
+
+from camunda.external_task.external_task_worker import ExternalTaskWorker
+
+ENGINE_URL = os.getenv('ENGINE_URL')
+
+
+def setup_worker(topic: str, handle_task: Callable, config: dict):
+    worker_id = f"{topic}_{uuid.uuid4().hex[:8]}"
+
+    ExternalTaskWorker(
+        worker_id=worker_id,
+        base_url=ENGINE_URL,
+        config=config,
+    ).subscribe([topic], handle_task)

external_workers/requirements.txt

@@ -3,7 +3,9 @@ pydantic==2.7.0
 web3==6.17.2
 ta==0.11.0
 redis==3.5.3
-
+pydantic-settings==2.2.1
+python-dotenv==1.0.1 ; python_version >= "3.10" and python_version < "4.0"
+eth_typing==4.1.0.
 boto3
 httpx
 openai

external_workers/testnet_arbitrage/get_pool_reserves.py

@@ -2,15 +2,15 @@
 
 from camunda.external_task.external_task import ExternalTask, TaskResult
 from camunda.external_task.external_task_worker import ExternalTaskWorker
-from testnet_arbitrage.execute_swap import get_pool_contract, get_pool_tokens, get_token_decimals, get_pool_reserves
+from external_workers.testnet_arbitrage.execute_swap import (
+    get_pool_contract,
+    get_pool_tokens,
+    get_token_decimals,
+    get_pool_reserves,
+)
 from web3 import Web3
 
-from config import (
-    WEB3_URL,
-    CAMUNDA_URL,
-    CAMUNDA_CLIENT_CONFIG,
-    TOPIC_NAME
-)
+from config import WEB3_URL, CAMUNDA_URL, CAMUNDA_CLIENT_CONFIG, TOPIC_NAME
 
 logger = logging.getLogger(__name__)
 
@@ -35,26 +35,71 @@ def handle_task(task: ExternalTask) -> TaskResult:
         else:
             raise ValueError("Token not in pool")
 
-        decimals = [get_token_decimals(pool_tokens[0]), get_token_decimals(pool_tokens[1])]
-        reserves = get_pool_reserves(pool_contract, target_token_address)
-        current_price = (
-            reserves.other_token_reserve / 10 ** decimals[1 - target_token_index]
-        ) / (reserves.target_token_reserve / 10 ** decimals[target_token_index])
-
+        decimals = [
+            get_token_decimals(pool_tokens[0]),
+            get_token_decimals(pool_tokens[1]),
+        ]
+        topic_name = task.get_topic_name()
+        if topic_name == 'uniswap_v2':  # TODO: Volynshchikov paste here needed topic
+            current_price = get_univ2_price(
+                pool_contract, target_token_address, decimals, target_token_index
+            )
+        elif topic_name == 'uniswap_v3':  # TODO: Volynshchikov paste here needed topic
+            current_price = get_univ3_price(
+                pool_contract, target_token_address, decimals, target_token_index
+            )
+        else:
+            return task.bpmn_error('unknown_amm', 'unknown_amm')
         variables['current_price'] = current_price
-        logger.info(f'''Success calculated price from pool - {target_token_address} 
-                        for token - {target_pool_address}''')
+        logger.info(
+            f'''Success calculated price from pool - {target_token_address} 
+                        for token - {target_pool_address}'''
+        )
         return task.complete(variables)
     except Exception as e:
         logger.error(e)
-        return task.failure(f'''Cannot get token price from pool - {target_token_address} 
+        return task.failure(
+            f'''Cannot get token price from pool - {target_token_address} 
                          for token - {target_pool_address}''',
-                     'Cannot resolve token price from pool',
-                     max_retries=3,
-                     retry_timeout=15000)
+            'Cannot resolve token price from pool',
+            max_retries=3,
+            retry_timeout=15000,
+        )
+
+
+def get_univ2_price(pool_contract, target_token_address, decimals, target_token_index):
+    reserves = get_pool_reserves(pool_contract, target_token_address)
+    current_price = (
+        reserves.other_token_reserve / 10 ** decimals[1 - target_token_index]
+    ) / (reserves.target_token_reserve / 10 ** decimals[target_token_index])
+    return current_price
+
+
+def get_univ3_price(pool_contract, target_token_address, decimals, target_token_index):
+    token_decimals_scalars = [10**d for d in decimals]
+    token_scalars_diff = token_decimals_scalars[1] / token_decimals_scalars[0]
+    values = pool_contract.functions.slot0().call()
+    names = [
+        "sqrtPriceX96",
+        "tick",
+        "observationIndex",
+        "observationCardinality",
+        "observationCardinalityNext",
+        "feeProtocol",
+        "unlocked",
+    ]
+    slot0 = dict(zip(names, values))
+    sqrt_price_x96 = slot0["sqrtPriceX96"]
+    price0 = ((sqrt_price_x96**2) / (2**192)) / token_scalars_diff
+    if target_token_index == 0:
+        return price0
+    else:
+        return 1 / price0
 
 
 if __name__ == "__main__":
     ExternalTaskWorker(
         worker_id="1", base_url=CAMUNDA_URL, config=CAMUNDA_CLIENT_CONFIG
-    ).subscribe([TOPIC_NAME], handle_task)
+    ).subscribe(
+        [TOPIC_NAME], handle_task
+    )  # TODO: Volynshchikov paste here needed topic

external_workers/testnet_arbitrage/pool_abi.json

@@ -709,5 +709,48 @@
     "payable": false,
     "stateMutability": "nonpayable",
     "type": "function"
+  },
+  {
+    "inputs": [],
+    "name": "slot0",
+    "outputs": [
+      {
+        "internalType": "uint160",
+        "name": "sqrtPriceX96",
+        "type": "uint160"
+      },
+      {
+        "internalType": "int24",
+        "name": "tick",
+        "type": "int24"
+      },
+      {
+        "internalType": "uint16",
+        "name": "observationIndex",
+        "type": "uint16"
+      },
+      {
+        "internalType": "uint16",
+        "name": "observationCardinality",
+        "type": "uint16"
+      },
+      {
+        "internalType": "uint16",
+        "name": "observationCardinalityNext",
+        "type": "uint16"
+      },
+      {
+        "internalType": "uint8",
+        "name": "feeProtocol",
+        "type": "uint8"
+      },
+      {
+        "internalType": "bool",
+        "name": "unlocked",
+        "type": "bool"
+      }
+    ],
+    "stateMutability": "view",
+    "type": "function"
   }
 ]

external_workers/testnet_arbitrage/uniswap_v3_execue_swap.py

@@ -0,0 +1,97 @@
+from pathlib import Path
+from time import time
+
+from pydantic import BaseModel, ValidationError
+from web3 import Web3
+from eth_account import Account
+
+from external_workers.common.utils import setup_worker
+from camunda.external_task.external_task import ExternalTask
+from config import PRIVATE_KEY, CAMUNDA_CLIENT_CONFIG
+
+TOPIC = 'executeSwapUniswapV3'
+VARIABLE_NOT_FOUND_ERR_CODE = "VARIABLE_NOT_FOUND"
+VARIABLE_NOT_FOUND_MSG = "Variable '{}' not found"
+
+
+with open(Path(__file__).parent / 'uniswap_v3_router_abi.json') as f:
+    UNISWAP_V3_ROUTER_ABI = f.read()
+
+
+class Vars(BaseModel):
+    rpc_url: str
+    uniswap_v3_router_address: str
+    token_in: str
+    token_out: str
+    amount_in: int
+    chain_id: int
+
+
+def handle_task(task: ExternalTask):
+    try:
+        variables = Vars.model_validate(task.get_variables())
+    except ValidationError as e:
+        return task.bpmn_error(VARIABLE_NOT_FOUND_ERR_CODE, e.json())
+    # Connect to Ethereum node via Infura or Alchemy
+    web3 = Web3(Web3.HTTPProvider(variables.rpc_url))
+
+    # Uniswap V3 Swap Router contract
+    uniswap_v3_router_address = variables.uniswap_v3_router_address
+
+    # Initialize the Uniswap V3 Router contract
+    uniswap_v3_router = web3.eth.contract(
+        address=web3.to_checksum_address(uniswap_v3_router_address),
+        abi=UNISWAP_V3_ROUTER_ABI,
+    )
+
+    # Tokens
+    token_in = variables.token_in
+    token_out = variables.token_out
+
+    fee = 10000  # Uniswap V3 pool fee (3000 = 0.3%)
+
+    # Wallet and account setup
+    private_key = PRIVATE_KEY
+    account = Account.from_key(private_key)
+    wallet_address = account.address
+
+    amount_in = variables.amount_in
+
+    # Gas price (use a proper provider like Etherscan to get current prices)
+    gas_price = web3.eth.gas_price
+
+    # Define the transaction parameters for swapExactInputSingle
+    params = {
+        'tokenIn': web3.to_checksum_address(token_in),
+        'tokenOut': web3.to_checksum_address(token_out),
+        'fee': fee,
+        'recipient': web3.to_checksum_address(wallet_address),
+        'deadline': time() + 600,  # 10 minutes from now
+        'amountIn': amount_in,
+        'amountOutMinimum': 0,  # set to 1 for simplicity, otherwise you should use Quoter for better estimation
+        'sqrtPriceLimitX96': 0,  # no limit on price slippage
+    }
+
+    # Build the transaction for the swap
+    swap_txn = uniswap_v3_router.functions.exactInputSingle(params).build_transaction(
+        {
+            'chainId': variables.chain_id,
+            'gas': 250000,
+            'gasPrice': gas_price,
+            'nonce': web3.eth.get_transaction_count(wallet_address),
+        }
+    )
+
+    # Sign the transaction
+    signed_txn = web3.eth.account.sign_transaction(swap_txn, private_key=private_key)
+
+    # Send the transaction
+    tx_hash = web3.eth.send_raw_transaction(signed_txn.rawTransaction)
+    receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
+    if receipt.status != 1:
+        raise ValueError
+    return task.complete({'tx_hash': tx_hash.hex()})
+
+
+if __name__ == '__main__':
+    setup_worker(TOPIC, handle_task, CAMUNDA_CLIENT_CONFIG)