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

Merge swap parsing into a shared function

Browse Source
This commit is contained in:
Luke Van Seters 2021-08-05 13:05:14 -04:00
parent 70514df025
commit 08a0784105
1 changed files with 29 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
mev_inspect/swaps.py
View File

@ -29,7 +29,7 @@ def get_swaps(traces: List[ClassifiedTrace]) -> List[Swap]:
elif trace.classification == Classification.swap: elif trace.classification == Classification.swap:
child_transfers = get_child_transfers(trace.trace_address, traces) child_transfers = get_child_transfers(trace.trace_address, traces)
swap = _build_swap( swap = _parse_swap(
trace, trace,
remove_inner_transfers(prior_transfers), remove_inner_transfers(prior_transfers),
remove_inner_transfers(child_transfers), remove_inner_transfers(child_transfers),
@ -41,94 +41,61 @@ def get_swaps(traces: List[ClassifiedTrace]) -> List[Swap]:
return swaps return swaps
def _build_swap( def _parse_swap(
trace: ClassifiedTrace, trace: ClassifiedTrace,
prior_transfers: List[Transfer], prior_transfers: List[Transfer],
child_transfers: List[Transfer], child_transfers: List[Transfer],
) -> Optional[Swap]: ) -> Optional[Swap]:
if trace.abi_name == UNISWAP_V2_PAIR_ABI_NAME: pool_address = trace.to_address
return _parse_uniswap_v2_swap(trace, prior_transfers, child_transfers) recipient_address = _get_recipient_address(trace)
elif trace.abi_name == UNISWAP_V3_POOL_ABI_NAME:
return _parse_uniswap_v3_swap(trace, child_transfers)
if recipient_address is None:
return None return None
transfers_to_pool = filter_transfers(prior_transfers, to_address=pool_address)
def _parse_uniswap_v3_swap( if len(transfers_to_pool) == 0:
trace: ClassifiedTrace, transfers_to_pool = filter_transfers(child_transfers, to_address=pool_address)
child_transfers: List[Transfer],
) -> Optional[Swap]: if len(transfers_to_pool) == 0:
pool_address = trace.to_address return None
recipient_address = (
transfers_from_pool_to_recipient = filter_transfers(
child_transfers, to_address=recipient_address, from_address=pool_address
)
if len(transfers_from_pool_to_recipient) != 1:
return None
transfer_in = transfers_to_pool[-1]
transfer_out = transfers_from_pool_to_recipient[0]
return Swap(
abi_name=trace.abi_name,
transaction_hash=trace.transaction_hash,
trace_address=trace.trace_address,
pool_address=pool_address,
from_address=transfer_in.from_address,
to_address=transfer_out.to_address,
token_in_address=transfer_in.token_address,
token_in_amount=transfer_in.amount,
token_out_address=transfer_out.token_address,
token_out_amount=transfer_out.amount,
)
def _get_recipient_address(trace: ClassifiedTrace) -> Optional[str]:
if trace.abi_name == UNISWAP_V3_POOL_ABI_NAME:
return (
trace.inputs["recipient"] trace.inputs["recipient"]
if trace.inputs is not None and "recipient" in trace.inputs if trace.inputs is not None and "recipient" in trace.inputs
else trace.from_address else trace.from_address
) )
elif trace.abi_name == UNISWAP_V2_PAIR_ABI_NAME:
transfers_to_pool = filter_transfers(child_transfers, to_address=pool_address) return (
transfers_from_pool_to_recipient = filter_transfers(
child_transfers, to_address=recipient_address, from_address=pool_address
)
if len(transfers_to_pool) == 0:
return None
if len(transfers_from_pool_to_recipient) != 1:
return None
transfer_in = transfers_to_pool[-1]
transfer_out = transfers_from_pool_to_recipient[0]
return Swap(
abi_name=UNISWAP_V3_POOL_ABI_NAME,
transaction_hash=trace.transaction_hash,
trace_address=trace.trace_address,
pool_address=pool_address,
from_address=transfer_in.from_address,
to_address=transfer_out.to_address,
token_in_address=transfer_in.token_address,
token_in_amount=transfer_in.amount,
token_out_address=transfer_out.token_address,
token_out_amount=transfer_out.amount,
)
def _parse_uniswap_v2_swap(
trace: ClassifiedTrace,
prior_transfers: List[Transfer],
child_transfers: List[Transfer],
) -> Optional[Swap]:
pool_address = trace.to_address
recipient_address = (
trace.inputs["to"] trace.inputs["to"]
if trace.inputs is not None and "to" in trace.inputs if trace.inputs is not None and "to" in trace.inputs
else trace.from_address else trace.from_address
) )
else:
transfers_to_pool = filter_transfers(prior_transfers, to_address=pool_address)
transfers_from_pool_to_recipient = filter_transfers(
child_transfers, to_address=recipient_address, from_address=pool_address
)
if len(transfers_to_pool) == 0:
return None return None
if len(transfers_from_pool_to_recipient) != 1:
return None
transfer_in = transfers_to_pool[-1]
transfer_out = transfers_from_pool_to_recipient[0]
return Swap(
abi_name=UNISWAP_V2_PAIR_ABI_NAME,
transaction_hash=trace.transaction_hash,
trace_address=trace.trace_address,
pool_address=pool_address,
from_address=transfer_in.from_address,
to_address=transfer_out.to_address,
token_in_address=transfer_in.token_address,
token_in_amount=transfer_in.amount,
token_out_address=transfer_out.token_address,
token_out_amount=transfer_out.amount,
)