crowetic/mev-inspect-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restructure classifier

Browse Source
This commit is contained in:
Gui Heise 2021-11-23 11:15:03 -05:00
parent d75e9b76ab
commit d7872db45c
2 changed files with 78 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
mev_inspect/classifiers/helpers.py
View File

@ -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

58
mev_inspect/classifiers/specs/zero_ex.py
View File

@ -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,