diff --git a/mev_inspect/inspectors/uniswap.py b/mev_inspect/inspectors/uniswap.py new file mode 100644 index 0000000..20aacda --- /dev/null +++ b/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 diff --git a/mev_inspect/processor.py b/mev_inspect/processor.py index d3eca33..6efa84d 100644 --- a/mev_inspect/processor.py +++ b/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: - self.base_provider = base_provider - self.inspectors = inspectors + def __init__(self, inspectors: List[Inspector]) -> None: + self._inspectors = inspectors - def get_transaction_evaluations(self, block_data): - for transaction_hash in block_data.transaction_hashes: - traces = block_data.get_filtered_traces(transaction_hash) - traces_json = [to_original_json_dict(trace) for trace in traces] + def get_transaction_evaluations( + self, + block: Block, + ) -> List[Classification]: + transaction_traces = ( + trace for trace in block.traces if trace.type != TraceType.reward + ) - for inspector in self.inspectors: - inspector.inspect(traces_json) + return [ + 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, + ) diff --git a/mev_inspect/schemas/classifications.py b/mev_inspect/schemas/classifications.py index 8a48e2f..3e0875f 100644 --- a/mev_inspect/schemas/classifications.py +++ b/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] diff --git a/testing_file.py b/testing_file.py index 0b392a8..8e824dd 100644 --- a/testing_file.py +++ b/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)