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

Remove old inspectors and add none return for 0 received

Browse Source
This commit is contained in:
Gui Heise 2022-01-18 14:27:59 -05:00
parent 172ed46b0b
commit f31e2525da
3 changed files with 10 additions and 195 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,
)

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

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

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