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

Support aave self-liquidations

Browse Source
This commit is contained in:
Gui Heise 2022-01-10 10:14:58 -05:00
parent 9894450e0c
commit 63087fc0e8
1 changed files with 19 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
mev_inspect/aave_liquidations.py
View File

@ -36,11 +36,11 @@ def get_aave_liquidations(
trace.transaction_hash, trace.trace_address, traces trace.transaction_hash, trace.trace_address, traces
) )
(debt_token_address, debt_purchase_amount) = _get_debt_data( (debt_token_address, debt_purchase_amount) = _get_debt_data(
child_traces, liquidator trace, child_traces, liquidator
) )
(received_token_address, received_amount) = _get_received_data( (received_token_address, received_amount) = _get_received_data(
child_traces, liquidator trace, child_traces, liquidator
) )
liquidations.append( liquidations.append(
@ -63,26 +63,28 @@ def get_aave_liquidations(
def _get_received_data( def _get_received_data(
child_traces: List[ClassifiedTrace], liquidator: str liquidation: 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"""
for child in child_traces: for child in child_traces:
child_transfer: Optional[Transfer] = get_transfer(child) child_transfer: Optional[Transfer] = get_transfer(child)
if child_transfer is not None: if child_transfer is not None and child_transfer.to_address == liquidator:
return child_transfer.token_address, child_transfer.amount
if child_transfer.to_address == liquidator: if liquidation.error is not None:
return liquidation.inputs["_collateral"], 0
return child_transfer.token_address, child_transfer.amount else:
raise RuntimeError(
raise RuntimeError("Transfer from AAVE to liquidator not found!") f"No payback or input data found for liquidation in tx: {liquidation.transaction_hash}"
)
def _get_debt_data( def _get_debt_data(
child_traces: List[ClassifiedTrace], liquidator: str liquidation: 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"""
@ -93,7 +95,12 @@ def _get_debt_data(
if child_transfer is not None: if child_transfer is not None:
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
raise RuntimeError("Transfer from liquidator to AAVE not found!") if liquidation.error is not None:
return liquidation.inputs["_reserve"], liquidation.inputs["_purchaseAmount"]
else:
raise RuntimeError(
f"No transfer or input data found for liquidation in tx: {liquidation.transaction_hash}"
)