diff --git a/mev_inspect/classifiers/specs/aave.py b/mev_inspect/classifiers/specs/aave.py index 854f057..e4cb959 100644 --- a/mev_inspect/classifiers/specs/aave.py +++ b/mev_inspect/classifiers/specs/aave.py @@ -1,13 +1,81 @@ +from typing import List, Optional, Tuple + from mev_inspect.schemas.classifiers import ( + Classifier, ClassifierSpec, DecodedCallTrace, - LiquidationClassifier, TransferClassifier, ) +from mev_inspect.schemas.liquidations import Liquidation from mev_inspect.schemas.traces import Protocol from mev_inspect.schemas.transfers import Transfer +class AaveLiquidationClassifier(Classifier): + def parse_liquidation( + self, liquidation_trace: DecodedCallTrace, child_transfers: List[Transfer] + ) -> Optional[Liquidation]: + + liquidator = liquidation_trace.from_address + + (debt_token_address, debt_purchase_amount) = self._get_debt_data( + liquidation_trace, child_transfers, liquidator + ) + + if debt_purchase_amount == 0: + return None + + (received_token_address, received_amount) = self._get_received_data( + liquidation_trace, child_transfers, liquidator + ) + + if received_amount == 0: + return None + + return Liquidation( + liquidated_user=liquidation_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=liquidation_trace.transaction_hash, + trace_address=liquidation_trace.trace_address, + block_number=liquidation_trace.block_number, + error=liquidation_trace.error, + ) + + def _get_received_data( + self, + liquidation_trace: DecodedCallTrace, + child_transfers: List[Transfer], + liquidator: str, + ) -> Tuple[str, int]: + + """Look for and return liquidator payback from liquidation""" + for transfer in child_transfers: + + if transfer.to_address == liquidator: + return transfer.token_address, transfer.amount + + return liquidation_trace.inputs["_collateral"], 0 + + def _get_debt_data( + self, + liquidation_trace: DecodedCallTrace, + child_transfers: List[Transfer], + liquidator: str, + ) -> Tuple[str, int]: + """Get transfer from liquidator to AAVE""" + + for transfer in child_transfers: + if transfer.from_address == liquidator: + return transfer.token_address, transfer.amount + + return liquidation_trace.inputs["_reserve"], 0 + + class AaveTransferClassifier(TransferClassifier): @staticmethod def get_transfer(trace: DecodedCallTrace) -> Transfer: @@ -26,7 +94,7 @@ AAVE_SPEC = ClassifierSpec( abi_name="AaveLendingPool", protocol=Protocol.aave, classifiers={ - "liquidationCall(address,address,address,uint256,bool)": LiquidationClassifier, + "liquidationCall(address,address,address,uint256,bool)": AaveLiquidationClassifier, }, ) @@ -35,8 +103,7 @@ ATOKENS_SPEC = ClassifierSpec( protocol=Protocol.aave, classifiers={ "transferOnLiquidation(address,address,uint256)": AaveTransferClassifier, - "transferFrom(address,address,uint256)": AaveTransferClassifier, }, ) -AAVE_CLASSIFIER_SPECS = [AAVE_SPEC, ATOKENS_SPEC] +AAVE_CLASSIFIER_SPECS: List[ClassifierSpec] = [AAVE_SPEC, ATOKENS_SPEC] diff --git a/mev_inspect/classifiers/specs/compound.py b/mev_inspect/classifiers/specs/compound.py index 039c165..3855d56 100644 --- a/mev_inspect/classifiers/specs/compound.py +++ b/mev_inspect/classifiers/specs/compound.py @@ -1,16 +1,100 @@ +from typing import List, Optional, Tuple + from mev_inspect.schemas.classifiers import ( + Classification, + ClassifiedTrace, + Classifier, ClassifierSpec, - LiquidationClassifier, + DecodedCallTrace, SeizeClassifier, ) +from mev_inspect.schemas.liquidations import Liquidation from mev_inspect.schemas.traces import Protocol +from mev_inspect.schemas.transfers import Transfer + + +class CompoundLiquidationClassifier(Classifier): + def parse_liquidation( + self, + liquidation_trace: DecodedCallTrace, + child_traces: List[ClassifiedTrace], + child_transfers: List[Transfer], + ) -> Optional[Liquidation]: + + seize_trace = self._get_seize_call(child_traces) + + if seize_trace is not None and seize_trace.inputs is not None: + + liquidator = seize_trace.inputs["liquidator"] + + (debt_token_address, debt_purchase_amount) = self._get_debt_data( + liquidator, child_transfers + ) + + if debt_purchase_amount == 0: + return None + + (received_token_address, received_amount) = self._get_received_data( + liquidator, child_transfers + ) + + if received_amount == 0: + return None + + return Liquidation( + liquidated_user=liquidation_trace.inputs["borrower"], + debt_token_address=debt_token_address, + liquidator_user=liquidator, + debt_purchase_amount=debt_purchase_amount, + protocol=liquidation_trace.protocol, + received_amount=received_amount, + received_token_address=received_token_address, + transaction_hash=liquidation_trace.transaction_hash, + trace_address=liquidation_trace.trace_address, + block_number=liquidation_trace.block_number, + error=liquidation_trace.error, + ) + return None + + def _get_seize_call( + self, 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 + + def _get_received_data( + self, liquidator: str, child_transfers: List[Transfer] + ) -> Tuple[str, int]: + """Look for and return payment for liquidation""" + + for transfer in child_transfers: + if transfer.to_address == liquidator: + return transfer.token_address, transfer.amount + + return liquidator, 0 + + def _get_debt_data( + self, liquidator: str, child_transfers: List[Transfer] + ) -> Tuple[str, int]: + """Get transfer from liquidator to compound""" + + for transfer in child_transfers: + + if transfer.from_address == liquidator: + return transfer.token_address, transfer.amount + + return liquidator, 0 + COMPOUND_V2_CETH_SPEC = ClassifierSpec( abi_name="CEther", protocol=Protocol.compound_v2, valid_contract_addresses=["0x4ddc2d193948926d02f9b1fe9e1daa0718270ed5"], classifiers={ - "liquidateBorrow(address,address)": LiquidationClassifier, + "liquidateBorrow(address,address)": CompoundLiquidationClassifier, "seize(address,address,uint256)": SeizeClassifier, }, ) @@ -20,7 +104,7 @@ CREAM_CETH_SPEC = ClassifierSpec( protocol=Protocol.cream, valid_contract_addresses=["0xD06527D5e56A3495252A528C4987003b712860eE"], classifiers={ - "liquidateBorrow(address,address)": LiquidationClassifier, + "liquidateBorrow(address,address)": CompoundLiquidationClassifier, "seize(address,address,uint256)": SeizeClassifier, }, ) @@ -48,7 +132,7 @@ COMPOUND_V2_CTOKEN_SPEC = ClassifierSpec( "0x80a2ae356fc9ef4305676f7a3e2ed04e12c33946", ], classifiers={ - "liquidateBorrow(address,uint256,address)": LiquidationClassifier, + "liquidateBorrow(address,uint256,address)": CompoundLiquidationClassifier, "seize(address,address,uint256)": SeizeClassifier, }, ) @@ -150,12 +234,12 @@ CREAM_CTOKEN_SPEC = ClassifierSpec( "0x58da9c9fc3eb30abbcbbab5ddabb1e6e2ef3d2ef", ], classifiers={ - "liquidateBorrow(address,uint256,address)": LiquidationClassifier, + "liquidateBorrow(address,uint256,address)": CompoundLiquidationClassifier, "seize(address,address,uint256)": SeizeClassifier, }, ) -COMPOUND_CLASSIFIER_SPECS = [ +COMPOUND_CLASSIFIER_SPECS: List[ClassifierSpec] = [ COMPOUND_V2_CETH_SPEC, COMPOUND_V2_CTOKEN_SPEC, CREAM_CETH_SPEC, diff --git a/mev_inspect/liquidations.py b/mev_inspect/liquidations.py index 88961f1..f0b6652 100644 --- a/mev_inspect/liquidations.py +++ b/mev_inspect/liquidations.py @@ -1,9 +1,12 @@ -from typing import List +from typing import List, Optional -from mev_inspect.aave_liquidations import get_aave_liquidations -from mev_inspect.compound_liquidations import get_compound_liquidations +from mev_inspect.classifiers.specs import get_classifier +from mev_inspect.schemas.classifiers import LiquidationClassifier from mev_inspect.schemas.liquidations import Liquidation -from mev_inspect.schemas.traces import Classification, ClassifiedTrace +from mev_inspect.schemas.traces import Classification, ClassifiedTrace, DecodedCallTrace +from mev_inspect.schemas.transfers import Transfer +from mev_inspect.traces import get_child_traces, is_child_trace_address +from mev_inspect.transfers import get_child_transfers def has_liquidations(classified_traces: List[ClassifiedTrace]) -> bool: @@ -14,9 +17,58 @@ def has_liquidations(classified_traces: List[ClassifiedTrace]) -> bool: return liquidations_exist -def get_liquidations( - classified_traces: List[ClassifiedTrace], -) -> List[Liquidation]: - aave_liquidations = get_aave_liquidations(classified_traces) - comp_liquidations = get_compound_liquidations(classified_traces) - return aave_liquidations + comp_liquidations +def get_liquidations(classified_traces: List[ClassifiedTrace]) -> List[Liquidation]: + + liquidations: List[Liquidation] = [] + parent_liquidations: List[DecodedCallTrace] = [] + + for trace in classified_traces: + + if not isinstance(trace, DecodedCallTrace): + continue + + if _is_child_liquidation(trace, parent_liquidations): + continue + + if trace.classification == Classification.liquidate: + + parent_liquidations.append(trace) + child_traces = get_child_traces( + trace.transaction_hash, trace.trace_address, classified_traces + ) + child_transfers = get_child_transfers( + trace.transaction_hash, trace.trace_address, child_traces + ) + liquidation = _parse_liquidation(trace, child_traces, child_transfers) + + if liquidation is not None: + liquidations.append(liquidation) + + return liquidations + + +def _parse_liquidation( + trace: DecodedCallTrace, + child_traces: List[ClassifiedTrace], + child_transfers: List[Transfer], +) -> Optional[Liquidation]: + + classifier = get_classifier(trace) + + if classifier is not None and issubclass(classifier, LiquidationClassifier): + return classifier.parse_liquidation(trace, child_transfers, child_traces) + return None + + +def _is_child_liquidation( + trace: DecodedCallTrace, parent_liquidations: List[DecodedCallTrace] +) -> bool: + + for parent in parent_liquidations: + if ( + trace.transaction_hash == parent.transaction_hash + and is_child_trace_address(trace.trace_address, parent.trace_address) + ): + return True + + return False diff --git a/mev_inspect/schemas/classifiers.py b/mev_inspect/schemas/classifiers.py index 043f1ff..4749f37 100644 --- a/mev_inspect/schemas/classifiers.py +++ b/mev_inspect/schemas/classifiers.py @@ -3,9 +3,10 @@ from typing import Dict, List, Optional, Type from pydantic import BaseModel +from .liquidations import Liquidation from .nft_trades import NftTrade from .swaps import Swap -from .traces import Classification, DecodedCallTrace, Protocol +from .traces import Classification, ClassifiedTrace, DecodedCallTrace, Protocol from .transfers import Transfer @@ -47,6 +48,15 @@ class LiquidationClassifier(Classifier): def get_classification() -> Classification: return Classification.liquidate + @staticmethod + @abstractmethod + def parse_liquidation( + trace: ClassifiedTrace, + child_transfers: List[Transfer], + child_traces: List[ClassifiedTrace] = [], + ) -> Optional[Liquidation]: + raise NotImplementedError() + class SeizeClassifier(Classifier): @staticmethod