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

Remove exceptions

Browse Source
This commit is contained in:
Gui Heise 2022-01-10 14:18:37 -05:00
parent 63087fc0e8
commit 2bb760874d
1 changed files with 18 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
mev_inspect/aave_liquidations.py
View File

@ -14,7 +14,7 @@ from mev_inspect.transfers import get_transfer
def get_aave_liquidations( def get_aave_liquidations(
traces: List[ClassifiedTrace], traces: List[ClassifiedTrace],
) -> List[Liquidation]: ) -> Optional[List[Liquidation]]:
"""Inspect list of classified traces and identify liquidation""" """Inspect list of classified traces and identify liquidation"""
liquidations: List[Liquidation] = [] liquidations: List[Liquidation] = []
@ -39,10 +39,16 @@ def get_aave_liquidations(
trace, child_traces, liquidator trace, child_traces, liquidator
) )
if debt_purchase_amount == 0:
continue
(received_token_address, received_amount) = _get_received_data( (received_token_address, received_amount) = _get_received_data(
trace, child_traces, liquidator trace, child_traces, liquidator
) )
if received_amount == 0:
continue
liquidations.append( liquidations.append(
Liquidation( Liquidation(
liquidated_user=trace.inputs["_user"], liquidated_user=trace.inputs["_user"],
@ -63,7 +69,9 @@ def get_aave_liquidations(
def _get_received_data( def _get_received_data(
liquidation: DecodedCallTrace, child_traces: List[ClassifiedTrace], liquidator: str liquidation_trace: DecodedCallTrace,
child_traces: List[ClassifiedTrace],
liquidator: str,
) -> Tuple[str, int]: ) -> Tuple[str, int]:
"""Look for and return liquidator payback from liquidation""" """Look for and return liquidator payback from liquidation"""
@ -74,17 +82,13 @@ def _get_received_data(
if child_transfer is not None and child_transfer.to_address == liquidator: if child_transfer is not None and child_transfer.to_address == liquidator:
return child_transfer.token_address, child_transfer.amount return child_transfer.token_address, child_transfer.amount
if liquidation.error is not None: return liquidation_trace.inputs["_collateral"], 0
return liquidation.inputs["_collateral"], 0
else:
raise RuntimeError(
f"No payback or input data found for liquidation in tx: {liquidation.transaction_hash}"
)
def _get_debt_data( def _get_debt_data(
liquidation: DecodedCallTrace, child_traces: List[ClassifiedTrace], liquidator: str liquidation_trace: DecodedCallTrace,
child_traces: List[ClassifiedTrace],
liquidator: str,
) -> Tuple[str, int]: ) -> Tuple[str, int]:
"""Get transfer from liquidator to AAVE""" """Get transfer from liquidator to AAVE"""
@ -97,10 +101,7 @@ def _get_debt_data(
if child_transfer.from_address == liquidator: if child_transfer.from_address == liquidator:
return child_transfer.token_address, child_transfer.amount return child_transfer.token_address, child_transfer.amount
if liquidation.error is not None: return (
return liquidation.inputs["_reserve"], liquidation.inputs["_purchaseAmount"] liquidation_trace.inputs["_reserve"],
liquidation_trace.inputs["_purchaseAmount"],
else: )
raise RuntimeError(
f"No transfer or input data found for liquidation in tx: {liquidation.transaction_hash}"
)