Merge pull request #138 from flashbots/cryptopunks-classifer

mev_inspect/classifiers/specs/__init__.py

@@ -11,6 +11,7 @@ from .weth import WETH_CLASSIFIER_SPECS, WETH_ADDRESS
 from .zero_ex import ZEROX_CLASSIFIER_SPECS
 from .balancer import BALANCER_CLASSIFIER_SPECS
 from .compound import COMPOUND_CLASSIFIER_SPECS
+from .cryptopunks import CRYPTOPUNKS_CLASSIFIER_SPECS
 
 ALL_CLASSIFIER_SPECS = (
     ERC20_CLASSIFIER_SPECS
@@ -21,6 +22,7 @@ ALL_CLASSIFIER_SPECS = (
     + ZEROX_CLASSIFIER_SPECS
     + BALANCER_CLASSIFIER_SPECS
     + COMPOUND_CLASSIFIER_SPECS
+    + CRYPTOPUNKS_CLASSIFIER_SPECS
 )
 
 _SPECS_BY_ABI_NAME_AND_PROTOCOL: Dict[

mev_inspect/classifiers/specs/cryptopunks.py

@@ -0,0 +1,31 @@
+from mev_inspect.schemas.traces import Protocol, Classification
+
+from mev_inspect.schemas.classifiers import (
+    ClassifierSpec,
+    Classifier,
+)
+
+
+class PunkBidAcceptanceClassifier(Classifier):
+    @staticmethod
+    def get_classification() -> Classification:
+        return Classification.punk_accept_bid
+
+
+class PunkBidClassifier(Classifier):
+    @staticmethod
+    def get_classification() -> Classification:
+        return Classification.punk_bid
+
+
+CRYPTO_PUNKS_SPEC = ClassifierSpec(
+    abi_name="cryptopunks",
+    protocol=Protocol.cryptopunks,
+    valid_contract_addresses=["0xb47e3cd837dDF8e4c57F05d70Ab865de6e193BBB"],
+    classifiers={
+        "enterBidForPunk(uint256)": PunkBidClassifier,
+        "acceptBidForPunk(uint256,uint256)": PunkBidAcceptanceClassifier,
+    },
+)
+
+CRYPTOPUNKS_CLASSIFIER_SPECS = [CRYPTO_PUNKS_SPEC]

mev_inspect/inspect_block.py

@@ -31,6 +31,7 @@ from mev_inspect.crud.liquidations import (
     write_liquidations,
 )
 from mev_inspect.miner_payments import get_miner_payments
+from mev_inspect.punks import get_punk_bid_acceptances, get_punk_bids, get_punk_snipes
 from mev_inspect.swaps import get_swaps
 from mev_inspect.transfers import get_transfers
 from mev_inspect.liquidations import get_liquidations
@@ -98,6 +99,12 @@ async def inspect_block(
     delete_liquidations_for_block(inspect_db_session, block_number)
     write_liquidations(inspect_db_session, liquidations)
 
+    punk_bids = get_punk_bids(classified_traces)
+    punk_bid_acceptances = get_punk_bid_acceptances(classified_traces)
+
+    punk_snipes = get_punk_snipes(punk_bids, punk_bid_acceptances)
+    logger.info(f"Block: {block_number} -- Found {len(punk_snipes)} punk snipes")
+
     miner_payments = get_miner_payments(
         block.miner, block.base_fee_per_gas, classified_traces, block.receipts
     )

mev_inspect/punks.py

@@ -0,0 +1,125 @@
+from typing import List, Optional
+from mev_inspect.schemas.traces import (
+    ClassifiedTrace,
+    Classification,
+    DecodedCallTrace,
+)
+from mev_inspect.schemas.punk_bid import PunkBid
+from mev_inspect.schemas.punk_accept_bid import PunkBidAcceptance
+from mev_inspect.schemas.punk_snipe import PunkSnipe
+from mev_inspect.traces import get_traces_by_transaction_hash
+
+
+def _get_highest_punk_bid_per_index(
+    punk_bids: List[PunkBid], punk_index: int
+) -> Optional[PunkBid]:
+    highest_punk_bid = None
+
+    for punk_bid in punk_bids:
+        if punk_bid.punk_index == punk_index:
+            if highest_punk_bid is None:
+                highest_punk_bid = punk_bid
+
+            elif punk_bid.price > highest_punk_bid.price:
+                highest_punk_bid = punk_bid
+
+    return highest_punk_bid
+
+
+def get_punk_snipes(
+    punk_bids: List[PunkBid], punk_bid_acceptances: List[PunkBidAcceptance]
+) -> List[PunkSnipe]:
+    punk_snipe_list = []
+
+    for punk_bid_acceptance in punk_bid_acceptances:
+        highest_punk_bid = _get_highest_punk_bid_per_index(
+            punk_bids, punk_bid_acceptance.punk_index
+        )
+
+        if highest_punk_bid is None:
+            continue
+
+        if highest_punk_bid.price > punk_bid_acceptance.min_price:
+            punk_snipe = PunkSnipe(
+                block_number=highest_punk_bid.block_number,
+                transaction_hash=highest_punk_bid.transaction_hash,
+                trace_address=highest_punk_bid.trace_address,
+                from_address=highest_punk_bid.from_address,
+                punk_index=highest_punk_bid.punk_index,
+                min_acceptance_price=punk_bid_acceptance.min_price,
+                acceptance_price=highest_punk_bid.price,
+            )
+
+            punk_snipe_list.append(punk_snipe)
+
+    return punk_snipe_list
+
+
+def get_punk_bid_acceptances(traces: List[ClassifiedTrace]) -> List[PunkBidAcceptance]:
+    punk_bid_acceptances = []
+
+    for _, transaction_traces in get_traces_by_transaction_hash(traces).items():
+        punk_bid_acceptances += _get_punk_bid_acceptances_for_transaction(
+            list(transaction_traces)
+        )
+
+    return punk_bid_acceptances
+
+
+def _get_punk_bid_acceptances_for_transaction(
+    traces: List[ClassifiedTrace],
+) -> List[PunkBidAcceptance]:
+    ordered_traces = list(sorted(traces, key=lambda t: t.trace_address))
+
+    punk_bid_acceptances = []
+
+    for trace in ordered_traces:
+        if not isinstance(trace, DecodedCallTrace):
+            continue
+
+        elif trace.classification == Classification.punk_accept_bid:
+            punk_accept_bid = PunkBidAcceptance(
+                block_number=trace.block_number,
+                transaction_hash=trace.transaction_hash,
+                trace_address=trace.trace_address,
+                from_address=trace.from_address,
+                punk_index=trace.inputs["punkIndex"],
+                min_price=trace.inputs["minPrice"],
+            )
+
+            punk_bid_acceptances.append(punk_accept_bid)
+
+    return punk_bid_acceptances
+
+
+def get_punk_bids(traces: List[ClassifiedTrace]) -> List[PunkBid]:
+    punk_bids = []
+
+    for _, transaction_traces in get_traces_by_transaction_hash(traces).items():
+        punk_bids += _get_punk_bids_for_transaction(list(transaction_traces))
+
+    return punk_bids
+
+
+def _get_punk_bids_for_transaction(traces: List[ClassifiedTrace]) -> List[PunkBid]:
+    ordered_traces = list(sorted(traces, key=lambda t: t.trace_address))
+
+    punk_bids = []
+
+    for trace in ordered_traces:
+        if not isinstance(trace, DecodedCallTrace):
+            continue
+
+        elif trace.classification == Classification.punk_bid:
+            punk_bid = PunkBid(
+                transaction_hash=trace.transaction_hash,
+                block_number=trace.block_number,
+                trace_address=trace.trace_address,
+                from_address=trace.from_address,
+                punk_index=trace.inputs["punkIndex"],
+                price=trace.value,
+            )
+
+            punk_bids.append(punk_bid)
+
+    return punk_bids

mev_inspect/schemas/punk_accept_bid.py

@@ -0,0 +1,12 @@
+from typing import List
+
+from pydantic import BaseModel
+
+
+class PunkBidAcceptance(BaseModel):
+    block_number: int
+    transaction_hash: str
+    trace_address: List[int]
+    from_address: str
+    punk_index: int
+    min_price: int

mev_inspect/schemas/punk_bid.py

@@ -0,0 +1,12 @@
+from typing import List
+
+from pydantic import BaseModel
+
+
+class PunkBid(BaseModel):
+    block_number: int
+    transaction_hash: str
+    trace_address: List[int]
+    from_address: str
+    punk_index: int
+    price: int

mev_inspect/schemas/punk_snipe.py

@@ -0,0 +1,13 @@
+from typing import List
+
+from pydantic import BaseModel
+
+
+class PunkSnipe(BaseModel):
+    block_number: int
+    transaction_hash: str
+    trace_address: List[int]
+    from_address: str
+    punk_index: int
+    min_acceptance_price: int
+    acceptance_price: int

mev_inspect/schemas/traces.py

@@ -31,6 +31,8 @@ class Classification(Enum):
     transfer = "transfer"
     liquidate = "liquidate"
     seize = "seize"
+    punk_bid = "punk_bid"
+    punk_accept_bid = "punk_accept_bid"
 
 
 class Protocol(Enum):
@@ -44,6 +46,7 @@ class Protocol(Enum):
     balancer_v1 = "balancer_v1"
     compound_v2 = "compound_v2"
     cream = "cream"
+    cryptopunks = "cryptopunks"
 
 
 class ClassifiedTrace(Trace):