120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
from typing import List, Optional
|
|
|
|
from mev_inspect.classifiers.helpers import get_debt_transfer, get_received_transfer
|
|
from mev_inspect.schemas.classifiers import (
|
|
ClassifiedTrace,
|
|
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(LiquidationClassifier):
|
|
@staticmethod
|
|
def parse_liquidation(
|
|
liquidation_trace: DecodedCallTrace,
|
|
child_transfers: List[Transfer],
|
|
child_traces: List[ClassifiedTrace],
|
|
) -> Optional[Liquidation]:
|
|
|
|
liquidator = liquidation_trace.from_address
|
|
liquidated = liquidation_trace.inputs["_user"]
|
|
|
|
debt_token_address = liquidation_trace.inputs["_reserve"]
|
|
received_token_address = liquidation_trace.inputs["_collateral"]
|
|
|
|
debt_purchase_amount = None
|
|
received_amount = None
|
|
|
|
debt_transfer = get_debt_transfer(liquidator, child_transfers)
|
|
|
|
received_transfer = get_received_transfer(liquidator, child_transfers)
|
|
|
|
if debt_transfer is not None and received_transfer is not None:
|
|
|
|
debt_token_address = debt_transfer.token_address
|
|
debt_purchase_amount = debt_transfer.amount
|
|
|
|
received_token_address = received_transfer.token_address
|
|
received_amount = received_transfer.amount
|
|
|
|
return Liquidation(
|
|
liquidated_user=liquidated,
|
|
debt_token_address=debt_token_address,
|
|
liquidator_user=liquidator,
|
|
debt_purchase_amount=debt_purchase_amount,
|
|
received_amount=received_amount,
|
|
received_token_address=received_token_address,
|
|
protocol=Protocol.aave,
|
|
transaction_hash=liquidation_trace.transaction_hash,
|
|
trace_address=liquidation_trace.trace_address,
|
|
block_number=liquidation_trace.block_number,
|
|
error=liquidation_trace.error,
|
|
)
|
|
|
|
else:
|
|
return None
|
|
|
|
|
|
class AaveTransferClassifier(TransferClassifier):
|
|
@staticmethod
|
|
def get_transfer(trace: DecodedCallTrace) -> Transfer:
|
|
return Transfer(
|
|
block_number=trace.block_number,
|
|
transaction_hash=trace.transaction_hash,
|
|
trace_address=trace.trace_address,
|
|
amount=trace.inputs["value"],
|
|
to_address=trace.inputs["to"],
|
|
from_address=trace.inputs["from"],
|
|
token_address=trace.to_address,
|
|
)
|
|
|
|
|
|
class AaveEthTransferClassifier(TransferClassifier):
|
|
@staticmethod
|
|
def get_transfer(trace: DecodedCallTrace) -> Transfer:
|
|
return Transfer(
|
|
block_number=trace.block_number,
|
|
transaction_hash=trace.transaction_hash,
|
|
trace_address=trace.trace_address,
|
|
amount=trace.inputs["_amount"],
|
|
to_address=trace.to_address,
|
|
from_address=trace.inputs["_user"],
|
|
token_address=trace.inputs["_reserve"],
|
|
)
|
|
|
|
|
|
AAVE_SPEC = ClassifierSpec(
|
|
abi_name="AaveLendingPool",
|
|
protocol=Protocol.aave,
|
|
classifiers={
|
|
"liquidationCall(address,address,address,uint256,bool)": AaveLiquidationClassifier,
|
|
},
|
|
)
|
|
|
|
AAVE_ETHTRANSFER_SPEC = ClassifierSpec(
|
|
abi_name="LendingPoolCore",
|
|
protocol=Protocol.aave,
|
|
classifiers={
|
|
"transferToReserve(address,address,uint256)": AaveEthTransferClassifier,
|
|
},
|
|
)
|
|
|
|
ATOKENS_SPEC = ClassifierSpec(
|
|
abi_name="aTokens",
|
|
protocol=Protocol.aave,
|
|
classifiers={
|
|
"transferOnLiquidation(address,address,uint256)": AaveTransferClassifier,
|
|
},
|
|
)
|
|
|
|
AAVE_CLASSIFIER_SPECS: List[ClassifierSpec] = [
|
|
AAVE_SPEC,
|
|
ATOKENS_SPEC,
|
|
AAVE_ETHTRANSFER_SPEC,
|
|
]
|