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

Tighten PR

Browse Source
This commit is contained in:
Gui Heise 2021-09-21 08:58:56 -04:00
parent 7f93466b35
commit e93a78b8ce
1 changed files with 6 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
mev_inspect/aave_liquidations.py
View File

@ -1,13 +1,12 @@
from typing import List
from mev_inspect.schemas.classified_traces import (
ClassifiedTrace,
DecodedCallTrace,
Classification,
)
from mev_inspect.schemas.liquidations import Liquidation
from mev_inspect.schemas.transfers import ERC20Transfer
LIQUIDATION_CONTRACT_ADDRESSES = [
"0x3dfd23A6c5E8BbcFc9581d2E864a68feb6a076d3",
@ -17,16 +16,12 @@ LIQUIDATION_CONTRACT_ADDRESSES = [
def find_liquidations_from_traces(
traces: List,
traces: List[Liquidation],
) -> List:
"""Inspect list of classified traces and identify liquidation"""
tx = []
seen_transactions = []
liquidations = []
result = []
unique_transfer_hashes: List[str]
transfers_to: List[List] = [[]]
transfers_from: List[List] = [[]]
liquidator: str
for trace in traces:
@ -35,12 +30,12 @@ def find_liquidations_from_traces(
# Check for liquidation and register trace and unique liquidator
if (
trace.classification == Classification.liquidate
and trace.transaction_hash not in tx
and trace.transaction_hash not in seen_transactions
):
liquidations.append(trace)
LIQUIDATION_CONTRACT_ADDRESSES.append(trace.from_address)
tx.append(trace.transaction_hash)
seen_transactions.append(trace.transaction_hash)
# Found liquidation, now parse inputs for data
for input in trace.inputs:
@ -66,65 +61,4 @@ def find_liquidations_from_traces(
)
)
elif is_transfer_from_liquidator(trace, unique_transfer_hashes):
# Add the transfer
liquidator = next(
(
LIQUIDATION_CONTRACT_ADDRESSES[i]
for i in range(len(LIQUIDATION_CONTRACT_ADDRESSES))
if trace.inputs["sender"] == LIQUIDATION_CONTRACT_ADDRESSES[i]
),
"",
)
transfers_from.append(
["from", liquidator, trace.transaction_hash, trace.inputs["amount"]]
)
unique_transfer_hashes.append(trace.transaction_hash)
elif is_transfer_to_liquidator(trace):
# Add the transfer
liquidator = next(
(
LIQUIDATION_CONTRACT_ADDRESSES[i]
for i in range(len(LIQUIDATION_CONTRACT_ADDRESSES))
if trace.inputs["recipient"]
== LIQUIDATION_CONTRACT_ADDRESSES[i]
),
"",
)
transfers_to.append(
["to", liquidator, trace.transaction_hash, trace.inputs["amount"]]
)
return [result, transfers_to, transfers_from]
def is_transfer_from_liquidator(
trace: ClassifiedTrace,
unique_transfer_hashes: List[str],
) -> bool:
"""Check if transfer is from liquidator"""
transfer = ERC20Transfer.from_trace(trace)
if (
trace.classification == Classification.transfer
and transfer.from_address in LIQUIDATION_CONTRACT_ADDRESSES
and trace.transaction_hash not in unique_transfer_hashes
):
return True
else:
return False
def is_transfer_to_liquidator(trace: ClassifiedTrace) -> bool:
"""Check if transfer is to liquidator"""
transfer = ERC20Transfer.from_trace(trace)
if (
trace.classification == Classification.transfer
and transfer.to_address in LIQUIDATION_CONTRACT_ADDRESSES
):
return True
else:
return False
return result