Multichain async

.gitignore

@@ -71,4 +71,5 @@ missing_events.json
 logs/*
 /token_details.csv
 /fastlane_bot/data/blockchain_data/*/token_detail/
-tx_log.txt
+missing_tokens_df.csv
+tokens_and_fee_df.csv

async_tests.py

@@ -0,0 +1,160 @@
+# coding=utf-8
+"""
+Contains the exchange class for UniswapV2. This class is responsible for handling UniswapV2 exchanges and updating the state of the pools.
+
+(c) Copyright Bprotocol foundation 2023.
+Licensed under MIT
+"""
+import asyncio
+import json
+import logging
+import os
+from dataclasses import dataclass
+from glob import glob
+from typing import List, Type, Tuple, Any, Dict
+
+import dotenv
+import pandas as pd
+import web3
+from web3 import Web3, AsyncWeb3
+from web3.contract import Contract, AsyncContract
+
+from fastlane_bot.data.abi import UNISWAP_V2_POOL_ABI
+from fastlane_bot.events.exchanges import exchange_factory
+from fastlane_bot.events.exchanges.base import Exchange
+from fastlane_bot.events.pools import pool_factory
+from fastlane_bot.events.pools.base import Pool
+
+dotenv.load_dotenv()
+
+logger = logging.getLogger(__name__)
+
+event_mappings = {}
+# Read Uniswap v2 event mappings and tokens
+uniswap_v2_filepath = "fastlane_bot/data/blockchain_data/ethereum/uniswap_v2_event_mappings.csv"
+uniswap_v2_event_mappings_df = pd.read_csv(uniswap_v2_filepath)
+event_mappings['uniswap_v2_pools'] = dict(
+    uniswap_v2_event_mappings_df[["address", "exchange"]].values
+)
+
+# Read Uniswap v3 event mappings and tokens
+uniswap_v3_filepath = "fastlane_bot/data/blockchain_data/ethereum/uniswap_v3_event_mappings.csv"
+uniswap_v3_event_mappings_df = pd.read_csv(uniswap_v3_filepath)
+event_mappings['uniswap_v3_pools'] = dict(
+    uniswap_v3_event_mappings_df[["address", "exchange"]].values
+)
+
+event_mappings['pancakeswap_v2_pools'] = dict(
+    uniswap_v2_event_mappings_df[["address", "exchange"]].values
+)
+
+event_mappings['pancakeswap_v3_pools'] = dict(
+    uniswap_v3_event_mappings_df[["address", "exchange"]].values
+)
+
+event_mappings['sushiswap_v2_pools'] = dict(
+    uniswap_v2_event_mappings_df[["address", "exchange"]].values
+)
+
+
+def exchange_name_from_event(event: Dict[str, Any]) -> str:
+    return next(
+        (
+            exchange_name
+            for exchange_name, pool_class in pool_factory._creators.items()
+            if pool_class.event_matches_format(event, event_mappings)
+        ),
+        None,
+    )
+
+async def get_token_and_fee(exchange_name, ex, address, contract, event):
+    try:
+        tkn0 = await ex.get_tkn0(
+            address, contract, event=event
+        )
+        tkn1 = await ex.get_tkn1(
+            address, contract, event=event
+        )
+        fee = await ex.get_fee(address, contract)
+        return exchange_name, address, tkn0, tkn1, fee
+    except Exception as e:
+        logger.info(f"\n\n failed [get_token_and_fee]: {e}, {ex}, {exchange_name}, {address}, {contract}, {event}")
+        return exchange_name, address, None, None, None
+
+
+async def main_get_tokens_and_fee(c):
+    vals = await asyncio.gather(
+        *[
+            get_token_and_fee(**args)
+            for args in c
+        ]
+    )
+    return pd.DataFrame(vals, columns=["exchange_name", "address", "tkn0_address", "tkn1_address", "fee"])
+
+w3_async = AsyncWeb3(
+    AsyncWeb3.AsyncHTTPProvider(f"https://eth-mainnet.alchemyapi.io/v2/{os.environ['WEB3_ALCHEMY_PROJECT_ID']}")
+)
+
+exchange_list = ('carbon_v1,bancor_v3,bancor_v2,bancor_pol,uniswap_v3,uniswap_v2,sushiswap_v2,balancer,pancakeswap_v2,'
+             'pancakeswap_v3')
+exchange_list = exchange_list.split(',')
+
+static_pool_data = pd.read_csv('fastlane_bot/data/blockchain_data/ethereum/static_pool_data.csv')
+static_pool_data["cid"] = [
+    Web3.keccak(text=f"{row['descr']}").hex()
+    for index, row in static_pool_data.iterrows()
+]
+
+with open('logs/20231106-013311/latest_event_data.json') as f:
+    latest_events = json.loads(f.read())
+
+abis = {}
+exchanges = {}
+for exchange in exchange_list:
+    exchanges[exchange] = exchange_factory.get_exchange(exchange)
+    abis[exchange] = exchanges[exchange].get_abi()
+
+
+contracts = []
+for event in latest_events:
+    exchange_name = exchange_name_from_event(event)
+    address = event['address']
+    if exchange_name == 'uniswap_v2':
+        contracts.append({
+            'exchange_name': exchange_name,
+            'ex': exchange_factory.get_exchange(exchange_name),
+            'address': address,
+            'contract': w3_async.eth.contract(address=address, abi=abis[exchange_name]),
+            'event': event
+        })
+
+# split contracts into chunks of 1000
+chunks = [contracts[i:i + 1000] for i in range(0, len(contracts), 1000)]
+dirname = 'temp'
+if not os.path.exists(dirname):
+    os.mkdir(dirname)
+
+for idx, chunk in enumerate(chunks):
+    print(idx, len(chunk))
+    loop = asyncio.get_event_loop()
+    tokens_and_fee_df = loop.run_until_complete(main_get_tokens_and_fee(chunk))
+    tokens_and_fee_df.to_csv(f"{dirname}/tokens_and_fee_df_{idx}.csv")
+
+### Above is working
+
+filepaths = glob(f"{dirname}/*.csv")
+pool_tokens_and_fee_df = pd.concat([pd.read_csv(filepath) for filepath in filepaths])
+pool_tokens_and_fee_df = pool_tokens_and_fee_df.drop_duplicates(subset=["exchange_name", "address"])
+pool_tokens_and_fee_df.to_csv("pool_tokens_and_fee_df.csv")
+#
+#
+# # clear temp dir
+# for filepath in filepaths:
+#     os.remove(filepath)
+#
+# # for each token in the pools, check whether we have the token info in the tokens.csv static data, and ifr not,
+# # add it
+# tokens = pool_tokens_and_fee_df["tkn0_address"].tolist() + pool_tokens_and_fee_df["tkn1_address"].tolist()
+# tokens = list(set(tokens))
+# print(f"tokens: {tokens[0]}")
+# tokens_df = pd.read_csv(f"fastlane_bot/data/blockchain_data/{mgr.blockchain}/tokens.csv")

fastlane_bot/bot.py

@@ -47,13 +47,13 @@
 import random
 import time
 from _decimal import Decimal
+from copy import copy
 from dataclasses import dataclass, asdict, field
 from datetime import datetime
 from typing import List, Dict, Tuple, Any, Callable
 from typing import Optional
 
 from web3 import Web3
-from web3.datastructures import AttributeDict
 
 from fastlane_bot.config import Config
 from fastlane_bot.helpers import (
@@ -66,8 +66,8 @@
     TxHelpers,
     TxHelpersBase,
     TradeInstruction,
-    Univ3Calculator,
     RouteStruct,
+    Univ3Calculator,
 )
 from fastlane_bot.helpers.routehandler import maximize_last_trade_per_tkn
 from fastlane_bot.tools.cpc import ConstantProductCurve as CPC, CPCContainer, T
@@ -78,10 +78,10 @@
 from .modes.pairwise_multi_bal import FindArbitrageMultiPairwiseBalancer
 from .modes.pairwise_multi_pol import FindArbitrageMultiPairwisePol
 from .modes.pairwise_single import FindArbitrageSinglePairwise
+from .modes.triangle_bancor_v3_two_hop import ArbitrageFinderTriangleBancor3TwoHop
 from .modes.triangle_multi import ArbitrageFinderTriangleMulti
 from .modes.triangle_single import ArbitrageFinderTriangleSingle
 from .modes.triangle_single_bancor3 import ArbitrageFinderTriangleSingleBancor3
-from .modes.triangle_bancor_v3_two_hop import ArbitrageFinderTriangleBancor3TwoHop
 from .utils import num_format, log_format, num_format_float
 
 
@@ -408,7 +408,7 @@ def _get_deadline(self, block_number) -> int:
             self.ConfigObj.w3.eth.block_number if block_number is None else block_number
         )
         return (
-            self.ConfigObj.w3.eth.getBlock(block_number).timestamp
+            self.ConfigObj.w3.eth.get_block(block_number).timestamp
             + self.ConfigObj.DEFAULT_BLOCKTIME_DEVIATION
         )
 
@@ -492,13 +492,19 @@ def _run(
 
         if data_validator:
             # Add random chance if we should check or not
+            arb_opp = copy(r)
             r = self.validate_optimizer_trades(
                 arb_opp=r, arb_mode=arb_mode, arb_finder=finder
             )
             if r is None:
-                self.ConfigObj.logger.info(
-                    "Math validation eliminated arb opportunity, restarting."
-                )
+                instructions_dic = arb_opp[2]
+                for instruction in instructions_dic:
+                    cid = instruction["cid"]
+                    # print exchange name for each cid
+                    pool = self.db.get_pool(cid=cid)
+                    self.ConfigObj.logger.info(
+                        f"\nMath validation eliminated arb opportunity, restarting. pool={pool}\n"
+                    )
                 return None
             if replay_mode:
                 pass
@@ -557,9 +563,9 @@ def validate_optimizer_trades(self, arb_opp, arb_mode, arb_finder):
                     return arb_opp
                     # pool_cid = pool_cid.split("-")[0]
                 cids.append(pool_cid)
-            if len(cids) > 3:
+            if len(cids) != 3:
                 self.ConfigObj.logger.info(
-                    f"Math validation not supported for more than 3 pools, returning to main flow."
+                    f"Math validation not supported for more than 3 pools, found {len(cids)} pools, returning to main flow."
                 )
                 return arb_opp
             max_trade_in = arb_finder.get_optimal_arb_trade_amts(
@@ -648,7 +654,9 @@ def validate_pool_data(self, arb_opp):
                         "Carbon pool not up to date, updating and restarting."
                     )
                     return False
-            elif current_pool.exchange_name in ["balancer",]:
+            elif current_pool.exchange_name in [
+                "balancer",
+            ]:
                 for idx, balance in enumerate(current_pool.token_balances):
                     if balance != fetched_pool[f"tkn{idx}_balance"]:
                         self.ConfigObj.logger.debug(
@@ -753,21 +761,39 @@ def calculate_profit(
         Tuple[Decimal, Decimal, Decimal]
             The updated best_profit, flt_per_bnt, and profit_usd.
         """
-        best_profit_fl_token = best_profit
+        best_profit_fl_token = Decimal(str(best_profit))
         if fl_token_with_weth != self.ConfigObj.WRAPPED_GAS_TOKEN_KEY:
             try:
-                fltkn_eth_conversion_rate = Decimal(str(CCm.bytknb(f"{self.ConfigObj.WRAPPED_GAS_TOKEN_KEY}").bytknq(f"{fl_token_with_weth}")[0].p))
+                fltkn_eth_conversion_rate = Decimal(
+                    str(
+                        CCm.bytknb(f"{self.ConfigObj.WRAPPED_GAS_TOKEN_KEY}")
+                        .bytknq(f"{fl_token_with_weth}")[0]
+                        .p
+                    )
+                )
                 best_profit_eth = best_profit_fl_token * fltkn_eth_conversion_rate
             except:
                 try:
-                    fltkn_eth_conversion_rate = 1/Decimal(str(CCm.bytknb(f"{fl_token_with_weth}").bytknq(f"{self.ConfigObj.WRAPPED_GAS_TOKEN_KEY}")[0].p))
+                    fltkn_eth_conversion_rate = 1 / Decimal(
+                        str(
+                            CCm.bytknb(f"{fl_token_with_weth}")
+                            .bytknq(f"{self.ConfigObj.WRAPPED_GAS_TOKEN_KEY}")[0]
+                            .p
+                        )
+                    )
                     best_profit_eth = best_profit_fl_token * fltkn_eth_conversion_rate
                 except Exception as e:
                     raise str(e)
         else:
             best_profit_eth = best_profit_fl_token
 
-        usd_eth_conversion_rate = Decimal(str(CCm.bypair(pair=f"{self.ConfigObj.WRAPPED_GAS_TOKEN_KEY}/{self.ConfigObj.STABLECOIN_KEY}")[0].p))
+        usd_eth_conversion_rate = Decimal(
+            str(
+                CCm.bypair(
+                    pair=f"{self.ConfigObj.WRAPPED_GAS_TOKEN_KEY}/{self.ConfigObj.STABLECOIN_KEY}"
+                )[0].p
+            )
+        )
         best_profit_usd = best_profit_eth * usd_eth_conversion_rate
         return best_profit_fl_token, best_profit_eth, best_profit_usd
 
@@ -957,7 +983,7 @@ def _handle_trade_instructions(
 
         # Get the flashloan amount and token address
         flashloan_amount = int(calculated_trade_instructions[0].amtin_wei)
-        flashloan_token_address = self.ConfigObj.w3.toChecksumAddress(
+        flashloan_token_address = self.ConfigObj.w3.to_checksum_address(
             self.db.get_token(key=fl_token).address
         )
 
@@ -1359,8 +1385,8 @@ def _ensure_connection(self, tenderly_fork: str):
         self.ConfigObj.w3 = Web3(Web3.HTTPProvider(tenderly_uri))
 
     def get_tokens_in_exchange(
-            self,
-            exchange_name: str,
+        self,
+        exchange_name: str,
     ) -> List[str]:
         """
         Gets all tokens that exist in pools on the specified exchange.