diff --git a/mev_inspect/classifiers/helpers.py b/mev_inspect/classifiers/helpers.py index 4e0e61b..9ce4c7f 100644 --- a/mev_inspect/classifiers/helpers.py +++ b/mev_inspect/classifiers/helpers.py @@ -1,4 +1,4 @@ -from typing import Optional, List, Sequence +from typing import Optional, List, Sequence, Tuple from mev_inspect.schemas.swaps import Swap from mev_inspect.schemas.transfers import Transfer, ETH_TOKEN_ADDRESS @@ -6,6 +6,17 @@ from mev_inspect.schemas.transfers import Transfer, ETH_TOKEN_ADDRESS from mev_inspect.schemas.traces import DecodedCallTrace, ClassifiedTrace +RFQ_SIGNATURES = [ + "fillRfqOrder((address,address,uint128,uint128,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128)", + "_fillRfqOrder((address,address,uint128,uint128,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128,address,bool,address)", +] +LIMIT_SIGNATURES = [ + "fillOrKillLimitOrder((address,address,uint128,uint128,uint128,address,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128)", + "fillLimitOrder((address,address,uint128,uint128,uint128,address,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128)", + "_fillLimitOrder((address,address,uint128,uint128,uint128,address,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128,address,address)", +] + + def create_swap_from_transfers( trace: DecodedCallTrace, recipient_address: str, @@ -86,19 +97,67 @@ def _filter_transfers( return filtered_transfers -def get_first_amount_transferred_to_address( - address: str, transfers: List[Transfer] +def is_valid_0x_swap( + trace: DecodedCallTrace, + child_transfers: List[Transfer], +) -> bool: + + # 1. There should be 2 child transfers, one for each settled leg of the order + if len(child_transfers) != 2: + raise ValueError( + f"A settled order should consist of 2 child transfers, not {len(child_transfers)}." + ) + + # 2. The function signature must be in the lists of supported signatures + if trace.function_signature not in (LIMIT_SIGNATURES + RFQ_SIGNATURES): + raise RuntimeError( + f"0x orderbook function {trace.function_signature} is not supported" + ) + + return True + + +def _get_taker_token_in_amount( + taker_address: str, token_in_address: str, child_transfers: List[Transfer] ) -> int: - for transfer in transfers: - if transfer.to_address == address: - return transfer.amount + + ANY_TAKER = "0x0000000000000000000000000000000000000000" + + if taker_address == ANY_TAKER: + for transfer in child_transfers: + if transfer.token_address == token_in_address: + return transfer.amount + else: + for transfer in child_transfers: + if transfer.to_address == taker_address: + return transfer.amount return 0 -def get_first_amount_transferred_by_token_address( - token_address: str, transfers: List[Transfer] -) -> int: - for transfer in transfers: - if transfer.token_address == token_address: - return transfer.amount - return 0 +def get_0x_token_in_data( + trace: DecodedCallTrace, child_transfers: List[Transfer] +) -> Tuple[str, int]: + + order: List = trace.inputs["order"] + token_in_address = order[0] + + if trace.function_signature in RFQ_SIGNATURES: + taker_address = order[5] + + elif trace.function_signature in LIMIT_SIGNATURES: + taker_address = order[6] + + token_in_amount = _get_taker_token_in_amount( + taker_address, token_in_address, child_transfers + ) + + return token_in_address, token_in_amount + + +def get_0x_token_out_data(trace: DecodedCallTrace) -> Tuple[str, int]: + + order: List = trace.inputs["order"] + token_out_address = order[1] + token_out_amount = trace.inputs["takerTokenFillAmount"] + + return token_out_address, token_out_amount diff --git a/mev_inspect/classifiers/specs/zero_ex.py b/mev_inspect/classifiers/specs/zero_ex.py index f865b54..b84de26 100644 --- a/mev_inspect/classifiers/specs/zero_ex.py +++ b/mev_inspect/classifiers/specs/zero_ex.py @@ -10,8 +10,9 @@ from mev_inspect.schemas.classifiers import ( SwapClassifier, ) from mev_inspect.classifiers.helpers import ( - get_first_amount_transferred_to_address, - get_first_amount_transferred_by_token_address, + is_valid_0x_swap, + get_0x_token_in_data, + get_0x_token_out_data, ) @@ -23,58 +24,11 @@ class ZeroExSwapClassifier(SwapClassifier): child_transfers: List[Transfer], ) -> Optional[Swap]: - ANY_TAKER = "0x0000000000000000000000000000000000000000" - RFQ_SIGNATURES = [ - "fillRfqOrder((address,address,uint128,uint128,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128)", - "_fillRfqOrder((address,address,uint128,uint128,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128,address,bool,address)", - ] - LIMIT_SIGNATURES = [ - "fillOrKillLimitOrder((address,address,uint128,uint128,uint128,address,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128)", - "fillLimitOrder((address,address,uint128,uint128,uint128,address,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128)", - "_fillLimitOrder((address,address,uint128,uint128,uint128,address,address,address,address,bytes32,uint64,uint256),(uint8,uint8,bytes32,bytes32),uint128,address,address)", - ] + assert is_valid_0x_swap(trace, child_transfers) - # Assumptions: - # 1. There should be 2 child transfers, one for each settled leg of the order - if len(child_transfers) != 2: - raise ValueError( - f"A settled order should consist of 2 child transfers, not {len(child_transfers)}." - ) + token_in_address, token_in_amount = get_0x_token_in_data(trace, child_transfers) - # 2. The function signature must be in the lists of supported signatures - if trace.function_signature not in (LIMIT_SIGNATURES + RFQ_SIGNATURES): - raise RuntimeError( - f"0x orderbook function {trace.function_signature} is not supported" - ) - - # The position of the token addresses and presence of takerTokenFillAmount - # is always ensured across order formats - - order: List = trace.inputs["order"] - - token_in_address: str = order[0] - - token_out_address: str = order[1] - token_out_amount = trace.inputs["takerTokenFillAmount"] - - # Find token in amount - if trace.function_signature in RFQ_SIGNATURES: - taker_address = order[5] - - elif trace.function_signature in LIMIT_SIGNATURES: - taker_address = order[6] - - token_out_amount = trace.inputs["takerTokenFillAmount"] - - if taker_address == ANY_TAKER: - token_in_amount = get_first_amount_transferred_by_token_address( - token_in_address, child_transfers - ) - - else: - token_in_amount = get_first_amount_transferred_to_address( - taker_address, child_transfers - ) + token_out_address, token_out_amount = get_0x_token_out_data(trace) return Swap( abi_name=trace.abi_name,