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

Remove unused and try

Browse Source
This commit is contained in:
Gui Heise 2021-09-24 15:53:46 -04:00
parent aedd6696b4
commit 8b5d1327a8
2 changed files with 11 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
mev_inspect/aave_liquidations.py
View File

@ -22,29 +22,27 @@ AAVE_CONTRACT_ADDRESSES: List[str] = [
] ]
def is_transfer_to_liquidator( def find_transfer_to_liquidator(
trace: ClassifiedTrace, liquidator: str trace: ClassifiedTrace, liquidator: str
) -> Optional[ClassifiedTrace]: ) -> Optional[ClassifiedTrace]:
"""Check if transfer is to liquidator""" """Check if transfer is to liquidator"""
if isinstance(trace, DecodedCallTrace): if isinstance(trace, DecodedCallTrace):
try:
if "recipient" in trace.inputs:
if ( if (
trace.inputs["recipient"] == liquidator trace.inputs["recipient"] == liquidator
and trace.from_address in AAVE_CONTRACT_ADDRESSES and trace.from_address in AAVE_CONTRACT_ADDRESSES
): ):
return trace return trace
except KeyError: elif "dst" in trace.inputs:
pass
try:
if ( if (
trace.inputs["dst"] == liquidator trace.inputs["dst"] == liquidator
and trace.from_address in AAVE_CONTRACT_ADDRESSES and trace.from_address in AAVE_CONTRACT_ADDRESSES
): ):
return trace return trace
except KeyError:
return None
return None return None
@ -54,7 +52,6 @@ def get_liquidations(
) -> List[Liquidation]: ) -> List[Liquidation]:
"""Inspect list of classified traces and identify liquidation""" """Inspect list of classified traces and identify liquidation"""
# liquidation_traces: List[DecodedCallTrace] = []
liquidations: List[Liquidation] = [] liquidations: List[Liquidation] = []
transfers_to: Dict = {} transfers_to: Dict = {}
unique_transaction_hashes: List = [] unique_transaction_hashes: List = []
@ -71,17 +68,18 @@ def get_liquidations(
unique_transaction_hashes.append(trace.transaction_hash) unique_transaction_hashes.append(trace.transaction_hash)
for t in traces: for t in traces:
to_result = is_transfer_to_liquidator(t, liquidator) to_result = find_transfer_to_liquidator(t, liquidator)
if to_result and not ( if to_result and not (
to_result.transaction_hash in transfers_to.keys() to_result.transaction_hash in transfers_to.keys()
): ):
transfers_to[trace.transaction_hash] = to_result transfers_to[trace.transaction_hash] = to_result
try: if "amount" in transfers_to[trace.transaction_hash].inputs:
received_amount = int( received_amount = int(
transfers_to[trace.transaction_hash].inputs["amount"] transfers_to[trace.transaction_hash].inputs["amount"]
) )
except KeyError:
elif "wad" in transfers_to[trace.transaction_hash].inputs:
received_amount = int( received_amount = int(
transfers_to[trace.transaction_hash].inputs["wad"] transfers_to[trace.transaction_hash].inputs["wad"]
) )
@ -101,5 +99,6 @@ def get_liquidations(
block_number=trace.block_number, block_number=trace.block_number,
) )
) )
print("\n")
print(liquidations)
return liquidations return liquidations

14
mev_inspect/schemas/abi.py
View File

@ -46,19 +46,5 @@ class ABIFunctionDescription(BaseModel):
return f"{self.name}({joined_input_types})" return f"{self.name}({joined_input_types})"
class ABIEventDescription(BaseModel):
type: Literal[ABIDescriptionType.event]
name: str
inputs: List[ABIDescriptionInput]
def get_selector(self) -> HexBytes:
signature = self.get_signature()
return Web3.sha3(text=signature)[0:4]
def get_signature(self) -> HexBytes:
joined_input_types = ",".join(input.type for input in self.inputs)
return f"{self.name}({joined_input_types})"
ABIDescription = Union[ABIFunctionDescription, ABIGenericDescription] ABIDescription = Union[ABIFunctionDescription, ABIGenericDescription]
ABI = List[ABIDescription] ABI = List[ABIDescription]