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

Merge pull request #235 from flashbots/liquidation-classifiers

Browse Source 
Add LiquidationClassifiers
This commit is contained in:
Gui Heise 2022-01-19 11:32:42 -05:00 committed by GitHub
commit e105ee4d29
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 285 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
mev_inspect/aave_liquidations.py
View File

@ -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,
)

27
mev_inspect/classifiers/helpers.py
View File

@ -1,9 +1,10 @@
from typing import List, Optional, Sequence
from mev_inspect.schemas.nft_trades import NftTrade
from mev_inspect.schemas.prices import ETH_TOKEN_ADDRESS
from mev_inspect.schemas.swaps import Swap
from mev_inspect.schemas.traces import ClassifiedTrace, DecodedCallTrace
from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS, Transfer
from mev_inspect.schemas.transfers import Transfer
def create_nft_trade_from_transfers(
@ -178,3 +179,27 @@ def _filter_transfers(
filtered_transfers.append(transfer)
return filtered_transfers
def get_received_transfer(
liquidator: str, child_transfers: List[Transfer]
) -> Optional[Transfer]:
"""Get transfer from AAVE to liquidator"""
for transfer in child_transfers:
if transfer.to_address == liquidator:
return transfer
return None
def get_debt_transfer(
liquidator: str, child_transfers: List[Transfer]
) -> Optional[Transfer]:
"""Get transfer from liquidator to AAVE"""
for transfer in child_transfers:
if transfer.from_address == liquidator:
return transfer
return None

2
mev_inspect/classifiers/specs/__init__.py
View File

@ -12,7 +12,7 @@ from .curve import CURVE_CLASSIFIER_SPECS
from .erc20 import ERC20_CLASSIFIER_SPECS
from .opensea import OPENSEA_CLASSIFIER_SPECS
from .uniswap import UNISWAP_CLASSIFIER_SPECS
from .weth import WETH_ADDRESS, WETH_CLASSIFIER_SPECS
from .weth import WETH_CLASSIFIER_SPECS
from .zero_ex import ZEROX_CLASSIFIER_SPECS
ALL_CLASSIFIER_SPECS = (

57
mev_inspect/classifiers/specs/aave.py
View File

@ -1,13 +1,65 @@
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,
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,
)
else:
return None
class AaveTransferClassifier(TransferClassifier):
@staticmethod
def get_transfer(trace: DecodedCallTrace) -> Transfer:
@ -26,7 +78,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 +87,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]

88
mev_inspect/classifiers/specs/compound.py
View File

@ -1,16 +1,86 @@
from typing import List, Optional
from mev_inspect.classifiers.helpers import get_debt_transfer, get_received_transfer
from mev_inspect.schemas.classifiers import (
Classification,
ClassifiedTrace,
ClassifierSpec,
DecodedCallTrace,
LiquidationClassifier,
SeizeClassifier,
)
from mev_inspect.schemas.liquidations import Liquidation
from mev_inspect.schemas.prices import CETH_TOKEN_ADDRESS
from mev_inspect.schemas.traces import Protocol
from mev_inspect.schemas.transfers import Transfer
class CompoundLiquidationClassifier(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["borrower"]
debt_token_address = liquidation_trace.to_address
received_token_address = liquidation_trace.inputs["cTokenCollateral"]
debt_purchase_amount = None
received_amount = None
debt_purchase_amount = (
liquidation_trace.value
if debt_token_address == CETH_TOKEN_ADDRESS and liquidation_trace.value != 0
else liquidation_trace.inputs["repayAmount"]
)
debt_transfer = get_debt_transfer(liquidator, child_transfers)
received_transfer = get_received_transfer(liquidator, child_transfers)
seize_trace = _get_seize_call(child_traces)
if debt_transfer is not None:
debt_token_address = debt_transfer.token_address
debt_purchase_amount = debt_transfer.amount
if received_transfer is not None:
received_token_address = received_transfer.token_address
received_amount = received_transfer.amount
elif seize_trace is not None and seize_trace.inputs is not None:
received_amount = seize_trace.inputs["seizeTokens"]
if received_amount is None:
return None
return Liquidation(
liquidated_user=liquidated,
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
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 +90,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 +118,7 @@ COMPOUND_V2_CTOKEN_SPEC = ClassifierSpec(
"0x80a2ae356fc9ef4305676f7a3e2ed04e12c33946",
],
classifiers={
"liquidateBorrow(address,uint256,address)": LiquidationClassifier,
"liquidateBorrow(address,uint256,address)": CompoundLiquidationClassifier,
"seize(address,address,uint256)": SeizeClassifier,
},
)
@ -150,14 +220,22 @@ 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,
CREAM_CTOKEN_SPEC,
]
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

5
mev_inspect/classifiers/specs/weth.py
View File

@ -3,6 +3,7 @@ from mev_inspect.schemas.classifiers import (
DecodedCallTrace,
TransferClassifier,
)
from mev_inspect.schemas.prices import WETH_TOKEN_ADDRESS
from mev_inspect.schemas.traces import Protocol
from mev_inspect.schemas.transfers import Transfer
@ -21,12 +22,10 @@ class WethTransferClassifier(TransferClassifier):
)
WETH_ADDRESS = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
WETH_SPEC = ClassifierSpec(
abi_name="WETH9",
protocol=Protocol.weth,
valid_contract_addresses=[WETH_ADDRESS],
valid_contract_addresses=[WETH_TOKEN_ADDRESS],
classifiers={
"transferFrom(address,address,uint256)": WethTransferClassifier,
"transfer(address,uint256)": WethTransferClassifier,

80
mev_inspect/compound_liquidations.py
View File

@ -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

72
mev_inspect/liquidations.py
View File

@ -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

12
mev_inspect/schemas/classifiers.py
View File

@ -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(
liquidation_trace: DecodedCallTrace,
child_transfers: List[Transfer],
child_traces: List[ClassifiedTrace],
) -> Optional[Liquidation]:
raise NotImplementedError()
class SeizeClassifier(Classifier):
@staticmethod

9
mev_inspect/schemas/prices.py
View File

@ -2,9 +2,8 @@ from datetime import datetime
from pydantic import BaseModel, validator
from mev_inspect.classifiers.specs.weth import WETH_ADDRESS
from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS
ETH_TOKEN_ADDRESS = "0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
WETH_TOKEN_ADDRESS = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
WBTC_TOKEN_ADDRESS = "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599"
LINK_TOKEN_ADDRESS = "0x514910771af9ca656af840dff83e8264ecf986ca"
YEARN_TOKEN_ADDRESS = "0x0bc529c00c6401aef6d220be8c6ea1667f6ad93e"
@ -20,7 +19,7 @@ CWBTC_TOKEN_ADDRESS = "0xc11b1268c1a384e55c48c2391d8d480264a3a7f4"
TOKEN_ADDRESSES = [
ETH_TOKEN_ADDRESS,
WETH_ADDRESS,
WETH_TOKEN_ADDRESS,
WBTC_TOKEN_ADDRESS,
LINK_TOKEN_ADDRESS,
YEARN_TOKEN_ADDRESS,
@ -36,7 +35,7 @@ TOKEN_ADDRESSES = [
]
COINGECKO_ID_BY_ADDRESS = {
WETH_ADDRESS: "weth",
WETH_TOKEN_ADDRESS: "weth",
ETH_TOKEN_ADDRESS: "ethereum",
WBTC_TOKEN_ADDRESS: "wrapped-bitcoin",
LINK_TOKEN_ADDRESS: "chainlink",

2
mev_inspect/schemas/transfers.py
View File

@ -2,8 +2,6 @@ from typing import List
from pydantic import BaseModel
ETH_TOKEN_ADDRESS = "0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
class Transfer(BaseModel):
block_number: int

3
mev_inspect/transfers.py
View File

@ -2,8 +2,9 @@ from typing import Dict, List, Optional, Sequence
from mev_inspect.classifiers.specs import get_classifier
from mev_inspect.schemas.classifiers import TransferClassifier
from mev_inspect.schemas.prices import ETH_TOKEN_ADDRESS
from mev_inspect.schemas.traces import ClassifiedTrace, DecodedCallTrace
from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS, Transfer
from mev_inspect.schemas.transfers import Transfer
from mev_inspect.traces import get_child_traces, is_child_trace_address

29
tests/liquidation_test.py → tests/test_aave.py
View File

@ -1,10 +1,10 @@
from typing import List
from mev_inspect.aave_liquidations import get_aave_liquidations
from mev_inspect.classifiers.trace import TraceClassifier
from mev_inspect.liquidations import get_liquidations
from mev_inspect.schemas.liquidations import Liquidation
from mev_inspect.schemas.prices import ETH_TOKEN_ADDRESS
from mev_inspect.schemas.traces import Protocol
from mev_inspect.transfers import ETH_TOKEN_ADDRESS
from tests.utils import load_test_block
@ -31,9 +31,10 @@ def test_single_weth_liquidation(trace_classifier: TraceClassifier):
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_aave_liquidations(classified_traces)
result = get_liquidations(classified_traces)
_assert_equal_list_of_liquidations(result, liquidations)
for liquidation in liquidations:
assert liquidation in result
def test_single_liquidation(trace_classifier: TraceClassifier):
@ -59,9 +60,10 @@ def test_single_liquidation(trace_classifier: TraceClassifier):
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_aave_liquidations(classified_traces)
result = get_liquidations(classified_traces)
_assert_equal_list_of_liquidations(result, liquidations)
for liquidation in liquidations:
assert liquidation in result
def test_single_liquidation_with_atoken_payback(trace_classifier: TraceClassifier):
@ -87,9 +89,10 @@ def test_single_liquidation_with_atoken_payback(trace_classifier: TraceClassifie
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_aave_liquidations(classified_traces)
result = get_liquidations(classified_traces)
_assert_equal_list_of_liquidations(result, liquidations)
for liquidation in liquidations:
assert liquidation in result
def test_multiple_liquidations_in_block(trace_classifier: TraceClassifier):
@ -139,10 +142,11 @@ def test_multiple_liquidations_in_block(trace_classifier: TraceClassifier):
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_aave_liquidations(classified_traces)
result = get_liquidations(classified_traces)
liquidations = [liquidation1, liquidation2, liquidation3]
_assert_equal_list_of_liquidations(result, liquidations)
for liquidation in liquidations:
assert liquidation in result
def test_liquidations_with_eth_transfer(trace_classifier: TraceClassifier):
@ -179,10 +183,11 @@ def test_liquidations_with_eth_transfer(trace_classifier: TraceClassifier):
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_aave_liquidations(classified_traces)
result = get_liquidations(classified_traces)
liquidations = [liquidation1, liquidation2]
_assert_equal_list_of_liquidations(result, liquidations)
for liquidation in liquidations:
assert liquidation in result
def _assert_equal_list_of_liquidations(

36
tests/test_compound.py
View File

@ -1,5 +1,5 @@
from mev_inspect.classifiers.trace import TraceClassifier
from mev_inspect.compound_liquidations import get_compound_liquidations
from mev_inspect.liquidations import get_liquidations
from mev_inspect.schemas.liquidations import Liquidation
from mev_inspect.schemas.traces import Protocol
from tests.utils import load_comp_markets, load_cream_markets, load_test_block
@ -30,8 +30,10 @@ def test_c_ether_liquidations(trace_classifier: TraceClassifier):
]
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_compound_liquidations(classified_traces)
assert result == liquidations
result = get_liquidations(classified_traces)
for liquidation in liquidations:
assert liquidation in result
block_number = 13207907
transaction_hash = (
@ -55,8 +57,10 @@ def test_c_ether_liquidations(trace_classifier: TraceClassifier):
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_compound_liquidations(classified_traces)
assert result == liquidations
result = get_liquidations(classified_traces)
for liquidation in liquidations:
assert liquidation in result
block_number = 13298725
transaction_hash = (
@ -79,8 +83,10 @@ def test_c_ether_liquidations(trace_classifier: TraceClassifier):
]
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_compound_liquidations(classified_traces)
assert result == liquidations
result = get_liquidations(classified_traces)
for liquidation in liquidations:
assert liquidation in result
def test_c_token_liquidation(trace_classifier: TraceClassifier):
@ -93,7 +99,7 @@ def test_c_token_liquidation(trace_classifier: TraceClassifier):
Liquidation(
liquidated_user="0xacdd5528c1c92b57045041b5278efa06cdade4d8",
liquidator_user="0xe0090ec6895c087a393f0e45f1f85098a6c33bef",
debt_token_address="0x39aa39c021dfbae8fac545936693ac917d5e7563",
debt_token_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
debt_purchase_amount=1207055531,
received_amount=21459623305,
received_token_address="0x70e36f6bf80a52b3b46b3af8e106cc0ed743e8e4",
@ -105,8 +111,10 @@ def test_c_token_liquidation(trace_classifier: TraceClassifier):
]
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_compound_liquidations(classified_traces)
assert result == liquidations
result = get_liquidations(classified_traces)
for liquidation in liquidations:
assert liquidation in result
def test_cream_token_liquidation(trace_classifier: TraceClassifier):
@ -119,7 +127,7 @@ def test_cream_token_liquidation(trace_classifier: TraceClassifier):
Liquidation(
liquidated_user="0x46bf9479dc569bc796b7050344845f6564d45fba",
liquidator_user="0xa2863cad9c318669660eb4eca8b3154b90fb4357",
debt_token_address="0x697256caa3ccafd62bb6d3aa1c7c5671786a5fd9",
debt_token_address="0x514910771af9ca656af840dff83e8264ecf986ca",
debt_purchase_amount=14857434973806369550,
received_amount=1547215810826,
received_token_address="0x44fbebd2f576670a6c33f6fc0b00aa8c5753b322",
@ -131,5 +139,7 @@ def test_cream_token_liquidation(trace_classifier: TraceClassifier):
]
block = load_test_block(block_number)
classified_traces = trace_classifier.classify(block.traces)
result = get_compound_liquidations(classified_traces)
assert result == liquidations
result = get_liquidations(classified_traces)
for liquidation in liquidations:
assert liquidation in result