wip feat: getting punk bids / accepts

mev_inspect/inspect_block.py

@@ -27,6 +27,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_bids
 from mev_inspect.swaps import get_swaps
 from mev_inspect.transfers import get_transfers
 from mev_inspect.liquidations import get_liquidations
@@ -91,6 +92,9 @@ 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)
+    logger.info(f"Block: {block_number} -- Found {len(punk_bids)} punk bids")
+
     miner_payments = get_miner_payments(
         block.miner, block.base_fee_per_gas, classified_traces, block.receipts
     )

mev_inspect/punks.py

@@ -0,0 +1,80 @@
+from typing import List
+
+from mev_inspect.schemas.traces import (
+    ClassifiedTrace,
+    Classification,
+    DecodedCallTrace,
+)
+from mev_inspect.schemas.punk_bid import Punk_Bid
+from mev_inspect.schemas.punk_accept_bid import Punk_Accept_Bid
+from mev_inspect.traces import get_traces_by_transaction_hash
+
+
+def get_punk_accept_bids(traces: List[ClassifiedTrace]) -> List[Punk_Accept_Bid]:
+    punk_accept_bids = []
+
+    for _, transaction_traces in get_traces_by_transaction_hash(traces).items():
+        punk_accept_bids += _get_punk_accept_bids_for_transaction(
+            list(transaction_traces)
+        )
+
+    return punk_accept_bids
+
+
+def _get_punk_accept_bids_for_transaction(
+    traces: List[ClassifiedTrace],
+) -> List[Punk_Accept_Bid]:
+    ordered_traces = list(sorted(traces, key=lambda t: t.trace_address))
+
+    punk_accept_bids = []
+
+    for trace in ordered_traces:
+        if not isinstance(trace, DecodedCallTrace):
+            continue
+
+        elif trace.classification == Classification.punk_accept_bid:
+            punk_accept_bid = Punk_Accept_Bid(
+                block_number=trace.block_number,
+                transaction_hash=trace.transaction_hash,
+                trace_address=trace.trace_address,
+                from_address=trace.from_address,
+                punk_index=trace.inputs["punk_index"],
+                min_price=trace.inputs["min_price"],
+            )
+
+            punk_accept_bids.append(punk_accept_bid)
+
+    return punk_accept_bids
+
+
+def get_punk_bids(traces: List[ClassifiedTrace]) -> List[Punk_Bid]:
+    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[Punk_Bid]:
+    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 = Punk_Bid(
+                transaction_hash=trace.transaction_hash,
+                block_number=trace.block_number,
+                trace_address=trace.trace_address,
+                from_address=trace.from_address,
+                punk_index=trace.inputs["punk_index"],
+                value=trace.value,
+            )
+
+            punk_bids.append(punk_bid)
+
+    return punk_bids