Support Inspector interface in processor. Support it in uniswap inspector

mev_inspect/inspectors/uniswap.py

@@ -0,0 +1,103 @@
+import json
+from typing import Optional
+
+from web3 import Web3
+
+from mev_inspect import utils
+from mev_inspect.config import load_config
+from mev_inspect.schemas.blocks import NestedTrace, TraceType
+from mev_inspect.schemas.classifications import Classification
+
+from .base import Inspector
+
+config = load_config()
+
+uniswap_router_abi = json.loads(config["ABI"]["UniswapV2Router"])
+uniswap_router_address = config["ADDRESSES"]["UniswapV2Router"]
+sushiswap_router_address = config["ADDRESSES"]["SushiswapV2Router"]
+
+uniswap_pair_abi = json.loads(config["ABI"]["UniswapV2Pair"])
+
+
+class UniswapInspector(Inspector):
+    def __init__(self, base_provider) -> None:
+        self.w3 = Web3(base_provider)
+
+        self.trading_functions = self.get_trading_functions()
+        self.uniswap_v2_router_contract = self.w3.eth.contract(
+            abi=uniswap_router_abi, address=uniswap_router_address
+        )
+        self.uniswap_router_trade_signatures = self.get_router_signatures()
+
+        self.uniswap_v2_pair_contract = self.w3.eth.contract(abi=uniswap_pair_abi)
+        self.uniswap_v2_pair_swap_signatures = (
+            self.uniswap_v2_pair_contract.functions.swap(
+                0, 0, uniswap_router_address, ""
+            ).selector
+        )  ## Note the address here doesn't matter, but it must be filled out
+        self.uniswap_v2_pair_reserves_signatures = (
+            self.uniswap_v2_pair_contract.functions.getReserves().selector
+        )  ## Called "checksigs" in mev-inpsect.ts
+
+        print("Built Uniswap inspector")
+
+    def get_trading_functions(self):
+        ## Gets all functions used for swapping
+        result = []
+
+        ## For each entry in the ABI
+        for abi in uniswap_router_abi:
+            ## Check to see if the entry is a function and if it is if the function's name starts with swap
+            if abi["type"] == "function" and abi["name"].startswith("swap"):
+                ## If so add it to our array
+                result.append(abi["name"])
+
+        return result
+
+    def get_router_signatures(self):
+        ## Gets the selector / function signatures of all the router swap functions
+        result = []
+
+        ## For each entry in the ABI
+        for abi in uniswap_router_abi:
+            ## Check to see if the entry is a function and if it is if the function's name starts with swap
+            if abi["type"] == "function" and abi["name"].startswith("swap"):
+                ## Add a parantheses
+                function = abi["name"] + "("
+
+                ## For each input in the function's input
+                for input in abi["inputs"]:
+
+                    ## Concat them into a string with commas
+                    function = function + input["internalType"] + ","
+
+                ## Take off the last comma, add a ')' to close the parentheses
+                function = function[:-1] + ")"
+
+                ## The result looks like this: 'swapETHForExactTokens(uint256,address[],address,uint256)'
+
+                ## Take the first 4 bytes of the sha3 hash of the above string.
+                selector = Web3.sha3(text=function)[0:4]
+
+                ## Add that to an array
+                result.append(selector)
+
+        return result
+
+    def inspect(self, nested_trace: NestedTrace) -> Optional[Classification]:
+        trace = nested_trace.trace
+
+        if (
+            trace.type == TraceType.call
+            and (
+                trace.action["to"] == uniswap_router_address.lower()
+                or trace.action["to"] == sushiswap_router_address.lower()
+            )
+            and utils.check_trace_for_signature(
+                trace, self.uniswap_router_trade_signatures
+            )
+        ):
+            # print("WIP, here is where there is a call that matches what we are looking for")
+            1 == 1
+
+        return None

mev_inspect/processor.py

@@ -1,15 +1,43 @@
-from mev_inspect.schemas.utils import to_original_json_dict
+from typing import List
+
+from mev_inspect.inspectors import Inspector
+from mev_inspect.schemas.blocks import Block, NestedTrace, TraceType
+from mev_inspect.schemas.classifications import (
+    Classification,
+    UnknownClassification,
+)
+from mev_inspect.traces import as_nested_traces
 
 
 class Processor:
-    def __init__(self, base_provider, inspectors) -> None:
+    def __init__(self, inspectors: List[Inspector]) -> None:
-        self.base_provider = base_provider
+        self._inspectors = inspectors
-        self.inspectors = inspectors
 
-    def get_transaction_evaluations(self, block_data):
+    def get_transaction_evaluations(
-        for transaction_hash in block_data.transaction_hashes:
+        self,
-            traces = block_data.get_filtered_traces(transaction_hash)
+        block: Block,
-            traces_json = [to_original_json_dict(trace) for trace in traces]
+    ) -> List[Classification]:
+        transaction_traces = (
+            trace for trace in block.traces if trace.type != TraceType.reward
+        )
 
-            for inspector in self.inspectors:
+        return [
-                inspector.inspect(traces_json)
+            self._run_inspectors(nested_trace)
+            for nested_trace in as_nested_traces(transaction_traces)
+        ]
+
+    def _run_inspectors(self, nested_trace: NestedTrace) -> Classification:
+        for inspector in self._inspectors:
+            classification = inspector.inspect(nested_trace)
+
+            if classification is not None:
+                return classification
+
+        internal_classifications = [
+            self._run_inspectors(subtrace) for subtrace in nested_trace.subtraces
+        ]
+
+        return UnknownClassification(
+            trace=nested_trace.trace,
+            internal_classifications=internal_classifications,
+        )

mev_inspect/schemas/classifications.py

@@ -1,5 +1,14 @@
+from typing import List
+
 from pydantic import BaseModel
 
+from .blocks import Trace
+
 
 class Classification(BaseModel):
     pass
+
+
+class UnknownClassification(Classification):
+    trace: Trace
+    internal_classifications: List[Classification]

testing_file.py

@@ -3,7 +3,7 @@ import argparse
 from web3 import Web3
 
 from mev_inspect import block
-from mev_inspect.inspector_uniswap import UniswapInspector
+from mev_inspect.inspectors.uniswap import UniswapInspector
 from mev_inspect.processor import Processor
 
 parser = argparse.ArgumentParser(description="Inspect some blocks.")
@@ -29,6 +29,6 @@ block_data = block.create_from_block_number(args.block_number[0], base_provider)
 uniswap_inspector = UniswapInspector(base_provider)
 
 ## Create a processor, pass in an ARRAY of inspects
-processor = Processor(base_provider, [uniswap_inspector, uniswap_inspector])
+processor = Processor([uniswap_inspector, uniswap_inspector])
 
 processor.get_transaction_evaluations(block_data)