Add transfer classifiers to classifier specs. Implement for erc20 and weth
This commit is contained in:
parent
f87016a561
commit
25e1cbcf83
@ -1,15 +1,34 @@
|
||||
from typing import Optional
|
||||
|
||||
from mev_inspect.schemas.classifier_specs import ClassifierSpec
|
||||
from mev_inspect.schemas.classified_traces import (
|
||||
Classification,
|
||||
DecodedCallTrace,
|
||||
)
|
||||
from mev_inspect.schemas.transfers import ERC20Transfer
|
||||
|
||||
|
||||
def classify_transfer(trace: DecodedCallTrace) -> Optional[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_SPEC = ClassifierSpec(
|
||||
abi_name="ERC20",
|
||||
transfer_classifiers={
|
||||
"transferFrom(address,address,uint256)": classify_transfer,
|
||||
"transfer(address,uint256)": classify_transfer,
|
||||
},
|
||||
classifications={
|
||||
"transferFrom(address,address,uint256)": Classification.transfer,
|
||||
"transfer(address,uint256)": Classification.transfer,
|
||||
"burn(address)": Classification.burn,
|
||||
},
|
||||
)
|
||||
|
||||
|
@ -1,17 +1,39 @@
|
||||
from typing import Optional
|
||||
|
||||
from mev_inspect.schemas.classifier_specs import ClassifierSpec
|
||||
from mev_inspect.schemas.classified_traces import (
|
||||
Classification,
|
||||
DecodedCallTrace,
|
||||
Protocol,
|
||||
)
|
||||
from mev_inspect.schemas.transfers import ERC20Transfer
|
||||
|
||||
|
||||
def classify_transfer(trace: DecodedCallTrace) -> Optional[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_SPEC = ClassifierSpec(
|
||||
abi_name="WETH9",
|
||||
protocol=Protocol.weth,
|
||||
valid_contract_addresses=["0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"],
|
||||
transfer_classifiers={
|
||||
"transferFrom(address,address,uint256)": classify_transfer,
|
||||
"transfer(address,uint256)": classify_transfer,
|
||||
},
|
||||
classifications={
|
||||
"transferFrom(address,address,uint256)": Classification.transfer,
|
||||
"transfer(address,uint256)": Classification.transfer,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
WETH_CLASSIFIER_SPECS = [WETH_SPEC]
|
||||
|
@ -3,13 +3,10 @@ from typing import Callable, Dict, List, Optional
|
||||
from pydantic import BaseModel
|
||||
|
||||
from .classified_traces import Classification, DecodedCallTrace, Protocol
|
||||
from .transfers import Transfer
|
||||
from .transfers import ERC20Transfer
|
||||
|
||||
|
||||
TransferClassifier = Callable[
|
||||
[DecodedCallTrace, List[DecodedCallTrace], List[DecodedCallTrace]],
|
||||
Transfer,
|
||||
]
|
||||
TransferClassifier = Callable[[DecodedCallTrace], Optional[ERC20Transfer]]
|
||||
|
||||
|
||||
class ClassifierSpec(BaseModel):
|
||||
|
Loading…
x
Reference in New Issue
Block a user