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

Add transfer parsing to transfer classifiers

Browse Source
This commit is contained in:
Luke Van Seters 2021-10-06 15:28:50 -04:00
parent 86ee26dd1a
commit 02c9c1cddc
3 changed files with 41 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
mev_inspect/classifiers/specs/erc20.py
View File

@ -1,14 +1,31 @@
from mev_inspect.schemas.classified_traces import DecodedCallTrace
from mev_inspect.schemas.classifiers import ( from mev_inspect.schemas.classifiers import (
ClassifierSpec, ClassifierSpec,
TransferClassifier, TransferClassifier,
) )
from mev_inspect.schemas.transfers import ERC20Transfer
def get_erc20_transfer(trace: DecodedCallTrace) -> ERC20Transfer:
return ERC20Transfer(
block_number=trace.block_number,
transaction_hash=trace.transaction_hash,
trace_address=trace.trace_address,
amount=trace.inputs["amount"],
to_address=trace.inputs["recipient"],
from_address=trace.inputs.get("sender", trace.from_address),
token_address=trace.to_address,
)
ERC20_TRANSFER_CLASSIFIER = TransferClassifier(get_transfer=get_erc20_transfer)
ERC20_SPEC = ClassifierSpec( ERC20_SPEC = ClassifierSpec(
abi_name="ERC20", abi_name="ERC20",
classifiers={ classifiers={
"transferFrom(address,address,uint256)": TransferClassifier(), "transferFrom(address,address,uint256)": ERC20_TRANSFER_CLASSIFIER,
"transfer(address,uint256)": TransferClassifier(), "transfer(address,uint256)": ERC20_TRANSFER_CLASSIFIER,
}, },
) )

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

@ -3,17 +3,33 @@ from mev_inspect.schemas.classified_traces import (
) )
from mev_inspect.schemas.classifiers import ( from mev_inspect.schemas.classifiers import (
ClassifierSpec, ClassifierSpec,
DecodedCallTrace,
TransferClassifier, TransferClassifier,
) )
from mev_inspect.schemas.transfers import ERC20Transfer
def get_weth_transfer(trace: DecodedCallTrace) -> ERC20Transfer:
return ERC20Transfer(
block_number=trace.block_number,
transaction_hash=trace.transaction_hash,
trace_address=trace.trace_address,
amount=trace.inputs["wad"],
to_address=trace.inputs["dst"],
from_address=trace.from_address,
token_address=trace.to_address,
)
WETH_TRANSFER_CLASSIFIER = TransferClassifier(get_transfer=get_weth_transfer)
WETH_SPEC = ClassifierSpec( WETH_SPEC = ClassifierSpec(
abi_name="WETH9", abi_name="WETH9",
protocol=Protocol.weth, protocol=Protocol.weth,
valid_contract_addresses=["0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"], valid_contract_addresses=["0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"],
classifiers={ classifiers={
"transferFrom(address,address,uint256)": TransferClassifier(), "transferFrom(address,address,uint256)": WETH_TRANSFER_CLASSIFIER,
"transfer(address,uint256)": TransferClassifier(), "transfer(address,uint256)": WETH_TRANSFER_CLASSIFIER,
}, },
) )

6
mev_inspect/schemas/classifiers.py
View File

@ -1,13 +1,15 @@
from typing import Dict, List, Optional, Union from typing import Callable, Dict, List, Optional, Union
from typing_extensions import Literal from typing_extensions import Literal
from pydantic import BaseModel from pydantic import BaseModel
from .classified_traces import Classification, Protocol from .classified_traces import Classification, DecodedCallTrace, Protocol
from .transfers import Transfer
class TransferClassifier(BaseModel): class TransferClassifier(BaseModel):
classification: Literal[Classification.transfer] = Classification.transfer classification: Literal[Classification.transfer] = Classification.transfer
get_transfer: Callable[[DecodedCallTrace], Transfer]
class SwapClassifier(BaseModel): class SwapClassifier(BaseModel):