diff --git a/mev_inspect/aave_liquidations.py b/mev_inspect/aave_liquidations.py deleted file mode 100644 index f72eca0..0000000 --- a/mev_inspect/aave_liquidations.py +++ /dev/null @@ -1,107 +0,0 @@ -from typing import List, Optional, Tuple - -from mev_inspect.schemas.liquidations import Liquidation -from mev_inspect.schemas.traces import ( - Classification, - ClassifiedTrace, - DecodedCallTrace, - Protocol, -) -from mev_inspect.schemas.transfers import Transfer -from mev_inspect.traces import get_child_traces, is_child_of_any_address -from mev_inspect.transfers import get_transfer - - -def get_aave_liquidations( - traces: List[ClassifiedTrace], -) -> List[Liquidation]: - - """Inspect list of classified traces and identify liquidation""" - liquidations: List[Liquidation] = [] - parent_liquidations: List[List[int]] = [] - - for trace in traces: - - if ( - trace.classification == Classification.liquidate - and isinstance(trace, DecodedCallTrace) - and not is_child_of_any_address(trace, parent_liquidations) - and trace.protocol == Protocol.aave - ): - - parent_liquidations.append(trace.trace_address) - liquidator = trace.from_address - - child_traces = get_child_traces( - trace.transaction_hash, trace.trace_address, traces - ) - (debt_token_address, debt_purchase_amount) = _get_debt_data( - trace, child_traces, liquidator - ) - - if debt_purchase_amount == 0: - continue - - (received_token_address, received_amount) = _get_received_data( - trace, child_traces, liquidator - ) - - if received_amount == 0: - continue - - liquidations.append( - Liquidation( - liquidated_user=trace.inputs["_user"], - debt_token_address=debt_token_address, - liquidator_user=liquidator, - debt_purchase_amount=debt_purchase_amount, - protocol=Protocol.aave, - received_amount=received_amount, - received_token_address=received_token_address, - transaction_hash=trace.transaction_hash, - trace_address=trace.trace_address, - block_number=trace.block_number, - error=trace.error, - ) - ) - - return liquidations - - -def _get_received_data( - liquidation_trace: DecodedCallTrace, - child_traces: List[ClassifiedTrace], - liquidator: str, -) -> Tuple[str, int]: - - """Look for and return liquidator payback from liquidation""" - for child in child_traces: - - child_transfer: Optional[Transfer] = get_transfer(child) - - if child_transfer is not None and child_transfer.to_address == liquidator: - return child_transfer.token_address, child_transfer.amount - - return liquidation_trace.inputs["_collateral"], 0 - - -def _get_debt_data( - liquidation_trace: DecodedCallTrace, - child_traces: List[ClassifiedTrace], - liquidator: str, -) -> Tuple[str, int]: - """Get transfer from liquidator to AAVE""" - - for child in child_traces: - - child_transfer: Optional[Transfer] = get_transfer(child) - - if child_transfer is not None: - - if child_transfer.from_address == liquidator: - return child_transfer.token_address, child_transfer.amount - - return ( - liquidation_trace.inputs["_reserve"], - 0, - ) diff --git a/mev_inspect/classifiers/specs/compound.py b/mev_inspect/classifiers/specs/compound.py index 2a2a896..76271ac 100644 --- a/mev_inspect/classifiers/specs/compound.py +++ b/mev_inspect/classifiers/specs/compound.py @@ -28,13 +28,9 @@ class CompoundLiquidationClassifier(LiquidationClassifier): received_token_address = liquidation_trace.inputs["cTokenCollateral"] received_amount = 0 - debt_transfer = _get_debt_transfer( - liquidator, child_transfers - ) + debt_transfer = _get_debt_transfer(liquidator, child_transfers) - received_transfer = _get_received_transfer( - liquidator, child_transfers - ) + received_transfer = _get_received_transfer(liquidator, child_transfers) seize_trace = _get_seize_call(child_traces) @@ -47,7 +43,10 @@ class CompoundLiquidationClassifier(LiquidationClassifier): received_amount = received_transfer.amount elif seize_trace is not None and seize_trace.inputs is not None: - received_amount = seize_trace.inputs['seizeTokens'] + received_amount = seize_trace.inputs["seizeTokens"] + + if received_amount == 0: + return None return Liquidation( liquidated_user=liquidated, @@ -231,6 +230,7 @@ def _get_seize_call(traces: List[ClassifiedTrace]) -> Optional[ClassifiedTrace]: return trace return None + def _get_received_transfer( liquidator: str, child_transfers: List[Transfer] ) -> Optional[Transfer]: @@ -241,7 +241,9 @@ def _get_received_transfer( return None -def _get_debt_transfer(liquidator: str, child_transfers: List[Transfer]) -> Optional[Transfer]: +def _get_debt_transfer( + liquidator: str, child_transfers: List[Transfer] +) -> Optional[Transfer]: """Get transfer from liquidator to compound""" for transfer in child_transfers: if transfer.from_address == liquidator: diff --git a/mev_inspect/compound_liquidations.py b/mev_inspect/compound_liquidations.py deleted file mode 100644 index b7fe667..0000000 --- a/mev_inspect/compound_liquidations.py +++ /dev/null @@ -1,80 +0,0 @@ -from typing import List, Optional - -from mev_inspect.schemas.liquidations import Liquidation -from mev_inspect.schemas.traces import Classification, ClassifiedTrace, Protocol -from mev_inspect.traces import get_child_traces - -V2_COMPTROLLER_ADDRESS = "0x3d9819210A31b4961b30EF54bE2aeD79B9c9Cd3B" -V2_C_ETHER = "0x4Ddc2D193948926D02f9B1fE9e1daa0718270ED5" -CREAM_COMPTROLLER_ADDRESS = "0x3d5BC3c8d13dcB8bF317092d84783c2697AE9258" -CREAM_CR_ETHER = "0xD06527D5e56A3495252A528C4987003b712860eE" - - -def get_compound_liquidations( - traces: List[ClassifiedTrace], -) -> List[Liquidation]: - - """Inspect list of classified traces and identify liquidation""" - liquidations: List[Liquidation] = [] - - for trace in traces: - if ( - trace.classification == Classification.liquidate - and ( - trace.protocol == Protocol.compound_v2 - or trace.protocol == Protocol.cream - ) - and trace.inputs is not None - and trace.to_address is not None - ): - # First, we look for cEther liquidations (position paid back via tx.value) - child_traces = get_child_traces( - trace.transaction_hash, trace.trace_address, traces - ) - seize_trace = _get_seize_call(child_traces) - - if seize_trace is not None and seize_trace.inputs is not None: - c_token_collateral = trace.inputs["cTokenCollateral"] - if trace.abi_name == "CEther": - liquidations.append( - Liquidation( - liquidated_user=trace.inputs["borrower"], - debt_token_address=trace.to_address, - liquidator_user=seize_trace.inputs["liquidator"], - debt_purchase_amount=trace.value, - protocol=trace.protocol, - received_amount=seize_trace.inputs["seizeTokens"], - received_token_address=c_token_collateral, - transaction_hash=trace.transaction_hash, - trace_address=trace.trace_address, - block_number=trace.block_number, - error=trace.error, - ) - ) - elif ( - trace.abi_name == "CToken" - ): # cToken liquidations where liquidator pays back via token transfer - liquidations.append( - Liquidation( - liquidated_user=trace.inputs["borrower"], - debt_token_address=trace.to_address, - liquidator_user=seize_trace.inputs["liquidator"], - debt_purchase_amount=trace.inputs["repayAmount"], - protocol=trace.protocol, - received_amount=seize_trace.inputs["seizeTokens"], - received_token_address=c_token_collateral, - transaction_hash=trace.transaction_hash, - trace_address=trace.trace_address, - block_number=trace.block_number, - error=trace.error, - ) - ) - return liquidations - - -def _get_seize_call(traces: List[ClassifiedTrace]) -> Optional[ClassifiedTrace]: - """Find the call to `seize` in the child traces (successful liquidation)""" - for trace in traces: - if trace.classification == Classification.seize: - return trace - return None