Create nft trade from transfers

mev_inspect/classifiers/helpers.py

@@ -1,9 +1,61 @@
 from typing import List, Optional, Sequence
+from mev_inspect.schemas.nft_trade import NftTrade
 
 from mev_inspect.schemas.swaps import Swap
 from mev_inspect.schemas.traces import ClassifiedTrace, DecodedCallTrace
 from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS, Transfer
 
+def create_nft_trade_from_transfers(
+    trace: DecodedCallTrace,
+    child_transfers: List[Transfer],
+    collection_address: str,
+    seller_address: str,
+    buyer_address: str,
+    exchange_wallet_address: Optional[str],
+) -> NftTrade:
+    transfers_to_buyer = _filter_transfers(
+        child_transfers, to_address=buyer_address
+    )
+    transfers_to_seller = _filter_transfers(
+        child_transfers, to_address=seller_address
+    )
+
+    if len(transfers_to_buyer) != 1 or len(transfers_to_seller) != 1:
+        return None
+    
+    if transfers_to_buyer[0].token_address != collection_address:
+        return None
+    
+    payment_token = transfers_to_seller[0].token_address
+    payment_amount = transfers_to_seller[0].amount
+    token_id = transfers_to_buyer[0].amount
+
+    if exchange_wallet_address is not None:
+        transfers_from_seller_to_exchange = _filter_transfers(
+            child_transfers, from_address=seller_address, to_address=exchange_wallet_address
+        )
+        transfers_from_buyer_to_exchange = _filter_transfers(
+            child_transfers, from_address=buyer_address, to_address=exchange_wallet_address
+        )
+        for fee in [*transfers_from_seller_to_exchange, *transfers_from_buyer_to_exchange]:
+            # Assumes that exchange fees are paid with the same token as the sale
+            payment_amount -= fee.amount
+
+    return NftTrade(
+        abi_name=trace.abi_name,
+        transaction_hash=trace.transaction_hash,
+        transaction_position=trace.transaction_position,
+        block_number=trace.block_number,
+        trace_address=trace.trace_address,
+        protocol=trace.protocol,
+        error=trace.error,
+        seller_address=seller_address,
+        buyer_address=buyer_address,
+        payment_token=payment_token,
+        payment_amount=payment_amount,
+        collection_address=collection_address,
+        token_id=token_id
+    )
 
 def create_swap_from_pool_transfers(
     trace: DecodedCallTrace,

mev_inspect/classifiers/specs/opensea.py

@@ -1,42 +1,30 @@
-from typing import List
+from typing import List, Optional
-from mev_inspect.classifiers.helpers import _filter_transfers
+from mev_inspect.classifiers.helpers import _filter_transfers, create_nft_trade_from_transfers
 from mev_inspect.schemas.classifiers import ClassifierSpec, NftTradeClassifier
 from mev_inspect.schemas.nft_trade import NftTrade
 from mev_inspect.schemas.traces import DecodedCallTrace, Protocol
 from mev_inspect.schemas.transfers import ETH_TOKEN_ADDRESS, Transfer
 
-OPENSEA_ETH_TOKEN_ADDRESS = "0x0000000000000000000000000000000000000000"
+OPENSEA_WALLET_ADDRESS = "0x5b3256965e7c3cf26e11fcaf296dfc8807c01073"
 
 class OpenseaClassifier(NftTradeClassifier):    
     @staticmethod
-    def parse_trade(trace: DecodedCallTrace) -> NftTrade:
+    def parse_trade(
-        uints = trace.inputs.get("uints")
+        trace: DecodedCallTrace,
+        child_transfers: List[Transfer],
+    ) -> Optional[NftTrade]:
         addresses = trace.inputs.get("addrs")
         buy_maker = addresses[1]
         sell_maker = addresses[8]
-        base_price = uints[4]
-        payment_token = addresses[6]
         target = addresses[4]
 
-        if payment_token == OPENSEA_ETH_TOKEN_ADDRESS:
+        return create_nft_trade_from_transfers(
-            # Opensea uses the zero-address as a sentinel value for Ether. Convert this
+            trace,
-            # to the normal eth token address.
+            child_transfers,
-            payment_token = ETH_TOKEN_ADDRESS
+            collection_address=target,
-
-        return NftTrade(
-            abi_name=trace.abi_name,
-            transaction_hash=trace.transaction_hash,
-            transaction_position=trace.transaction_position,
-            block_number=trace.block_number,
-            trace_address=trace.trace_address,
-            protocol=trace.protocol,
-            error=trace.error,
             seller_address=sell_maker,
             buyer_address=buy_maker,
-            payment_token=payment_token,
+            exchange_wallet_address=OPENSEA_WALLET_ADDRESS,
-            payment_amount=base_price,
-            collection_address=target,
-            token_uri=0 # Todo
         )
 
 

mev_inspect/nft_trades.py

@@ -5,6 +5,10 @@ from mev_inspect.schemas.nft_trade import NftTrade
 from mev_inspect.schemas.traces import Classification, ClassifiedTrace, DecodedCallTrace
 from mev_inspect.schemas.transfers import Transfer
 from mev_inspect.traces import get_traces_by_transaction_hash
+from mev_inspect.transfers import (
+    get_child_transfers,
+    remove_child_transfers_of_transfers,
+)
 
 def get_nft_trades(traces: List[ClassifiedTrace]) -> List[NftTrade]:
     nft_trades = []
@@ -29,16 +33,28 @@ def _get_nft_trades_for_transaction(
             continue
 
         elif trace.classification == Classification.nft_trade:
-            nft_transfer = _parse_trade(trace)
+            child_transfers = get_child_transfers(
+                trace.transaction_hash,
+                trace.trace_address,
+                traces,
+            )
+            nft_transfer = _parse_trade(
+                trace,
+                remove_child_transfers_of_transfers(child_transfers),
+            )
             
+            if nft_transfer is not None:
                 nft_trades.append(nft_transfer)
 
     return nft_trades
 
-def _parse_trade(trace: DecodedCallTrace) -> Optional[NftTrade]:
+def _parse_trade(
+    trace: DecodedCallTrace,
+    child_transfers: List[Transfer],
+) -> Optional[NftTrade]:
     classifier = get_classifier(trace)
 
     if classifier is not None and issubclass(classifier, NftTradeClassifier):
-        return classifier.parse_trade(trace)
+        return classifier.parse_trade(trace, child_transfers)
 
     return None

mev_inspect/schemas/classifiers.py

@@ -60,7 +60,10 @@ class NftTradeClassifier(Classifier):
 
     @staticmethod
     @abstractmethod
-    def parse_trade(trace: DecodedCallTrace) -> NftTrade:
+    def parse_trade(
+        trace: DecodedCallTrace,
+        child_transfers: List[Transfer],
+    ) -> Optional[NftTrade]:
         return NotImplementedError()
 
 

mev_inspect/schemas/nft_trade.py

@@ -17,4 +17,4 @@ class NftTrade(BaseModel):
     payment_token: str
     payment_amount: int
     collection_address: str
-    token_uri: int
+    token_id: int