fixed inspector functionality, added strategy data type and aave address provider

liquidation_inspector.py

@@ -8,59 +8,73 @@ from typing import List, Optional
 from mev_inspect import utils
 from mev_inspect.config import load_config
 from mev_inspect.schemas.blocks import NestedTrace, TraceType
-from mev_inspect.schemas.classifications import Classification
+from mev_inspect.schemas.classified_traces import Classification, ClassifiedTrace
-from mev_inspect.schemas.strategy import Strategy, Liquidation
+from mev_inspect.schemas.strategy import StrategyType, Strategy, Liquidation
+from mev_inspect.classifier_specs import CLASSIFIER_SPECS
+from mev_inspect.trace_classifier import TraceClassifier
+from mev_inspect import block
 
-from .base import StrategyInspector
+# poetry run inspect -b 12498502 -r 'http://162.55.96.141:8546/'
+result = []
 
-class LiquidationInspector(StrategyInspector):
+# Inspect list of classified traces and identify liquidation
-
+def liquidations(traces: List[ClassifiedTrace]):
-	def __init__(self):
+	event = []
-		self.result = []
+	# For each trace
-
+	for i in range(1, len(traces)):
-	# Inspect list of classified traces and identify liquidation
+		trace = traces[i]
-	def inspect(self, traces):
+		try:
-
-		event = []
-
-		# For each trace
-		for i in range(1, len(traces)):
-			trace = traces[i]
 			next = traces[i+1]
-
+		except IndexError:
-			# Liquidation condition
+			break
-			if trace.classification == Classification.liquidate:
+		# Liquidation condition
-
+		if trace.classification == Classification.liquidate:
-				# Collateral data from the liquidation.
+		# Collateral data from the liquidation.
-				# The inputs will differ by DEX, this is AAVE
+		# The inputs will differ by DEX, this is AAVE
-
+			liquidator = trace.from_address
-				for i in trace.inputs:
+			prev = traces[i-1]
-					if(i["name"] == '_purchaseAmount'):
+			#print(f"Previous: {prev.classification} from {prev.from_address} to {prev.to_address}")
-						liquidation_amount = trace.inputs[i]
+			print(f"Liquidation found: {liquidator}")
-					elif (i["name"] == '_collateral'):
+			print(f"Hash: {trace.transaction_hash}")
-						collateral_type = trace.inputs[i]
+			for i in trace.inputs:
-					elif (i["name"] == '_reserve'):
+				if(i == '_purchaseAmount'):
-						collateral_source = trace.inputs[i]
+					liquidation_amount = trace.inputs[i]
-
+					print(f"\tAmount: {liquidation_amount}")
-
+				elif (i == '_collateral'):
+					collateral_type = trace.inputs[i]
+					print(f"\tType: {collateral_type}")
+				elif (i == '_reserve'):
+					underlying = trace.inputs[i]
+					print(f"\tUnderlying: {underlying}")
+				elif(i == '_user'):
+					liquidated_usr = trace.inputs[i]
+					print(f"\tLiquidated: {liquidated_usr}")
 				# Define the address of the liquidator
-				liquidator = trace.from_address
 
 				# Find a transfer before liquidation with a to_address corresponding to the liquidator
-				for tx in traces:
+			for tx in traces:
-					if (tx.classification == Classification.transfer & tx.to_address == liquidator):
+				if ((tx.classification==Classification.transfer) and (tx.inputs['recipient'] == liquidator)):
-						#Calculate profit
+					amount_received = tx.inputs['amount']
-						amount_sent = tx.value
+					print(f"Transfer to liquidator {liquidator}: \nAmount in received token: {tx.inputs['amount']} \nTransaction: {tx.transaction_hash}")
-						amount_received = next.value
-						profit = amount_received - amount_sent
-						# Add gas used*price and coinbase check
 
+				#Calculate profit
+				else:
+					# Add gas used*price and coinbase check
+			# Tag liquidation
+			result.append(Liquidation(strategy=StrategyType.liquidation,
+									  traces=[trace, next],
+									  protocols=[trace.protocol],
+									  collateral_type=collateral_type,
+									  collateral_amount=liquidation_amount,
+									  underlying=underlying,
+									  collateral_source="",
+									  profit=profit))
+	return result
 
-
+rpc = 'http://162.55.96.141:8546/'
-				# Tag liquidation
+block_number = 12498502
-				event.append(Strategy(strategy=StategyType.liquidation,
+base_provider = Web3.HTTPProvider(rpc)
-									  traces=[trace],
+block_data = block.create_from_block_number(block_number, base_provider)
-									  protocols=[trace.protocol]))
+trace_clasifier = TraceClassifier(CLASSIFIER_SPECS)
-
+classified_traces = trace_clasifier.classify(block_data.traces)
-
+print(liquidations(classified_traces)[2])
-		return result

mev_inspect/classifier_specs.py

@@ -99,19 +99,27 @@ ERC20_SPEC = ClassifierSpec(
     },
 )
 
-AAVE_SPEC = ClassifierSpec(
+AAVE_CONTRACT_SPECS = [
-    abi_name="AaveLendingPool",
+    ClassifierSpec(
-    protocol= Protocol.aave,
+        abi_name="AaveLendingPool",
-    classifications={
+        protocol= Protocol.aave,
-        "liquidationCall(address,address,address,uint256,bool)": Classification.liquidate,
+        classifications={
-    },
+            "liquidationCall(address,address,address,uint256,bool)": Classification.liquidate,},
-)
+    ),
+    ClassifierSpec(
+        abi_name="AaveLendingPoolAddressProvider",
+        protocol= Protocol.aave,
+        valid_contract_addresses=['0x24a42fD28C976A61Df5D00D0599C34c4f90748c8'],
+    ),
+    ]
+
+
 
 
 CLASSIFIER_SPECS = [
     *UNISWAP_V3_CONTRACT_SPECS,
     *UNISWAPPY_V2_CONTRACT_SPECS,
-    AAVE_SPEC,
+    *AAVE_CONTRACT_SPECS,
     ERC20_SPEC,
     UNISWAP_V3_POOL_SPEC,
     UNISWAPPY_V2_PAIR_SPEC,

mev_inspect/schemas/strategy.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, List
 from enum import Enum
 from pydantic import BaseModel
 from .classified_traces import ClassifiedTrace, Protocol
@@ -10,13 +10,10 @@ class StrategyType(Enum):
 class Strategy(BaseModel):
 	strategy: StrategyType
 	traces: List[ClassifiedTrace]
-	protocols:	List[Protocol]
+	protocols: List[Protocol]
 
 class Liquidation(Strategy):
     collateral_type: str
     collateral_amount: int
     collateral_source: str
-    profit: int
+    underlying: str
-
-class Arbitrage(Strategy):
-#	strategy: Literal[StrategyType.arbitrage]

mev_inspect/strat/liquidation_inspector.py

@@ -0,0 +1,61 @@
+import json
+from typing import Optional
+
+from web3 import Web3
+
+from typing import List, Optional
+
+from mev_inspect import utils
+from mev_inspect.config import load_config
+from mev_inspect.schemas.blocks import NestedTrace, TraceType
+from mev_inspect.schemas.classifications import Classification
+from mev_inspect.schemas.strategy import Strategy, Liquidation
+
+from .base import StrategyInspector
+
+class LiquidationInspector(StrategyInspector):
+
+	def __init__(self):
+		self.result = []
+
+	# Inspect list of classified traces and identify liquidation
+	def inspect(self, traces):
+
+		event = []
+
+		# For each trace
+		for i in range(1, len(traces)):
+			trace = traces[i]
+
+			# Liquidation condition
+			if trace.classification == Classification.liquidate:
+
+				# Collateral data from the liquidation.
+				# The inputs will differ by DEX, this is AAVE
+
+				for i in trace.inputs:
+					if( i["name"] == '_purchaseAmount'):
+						value = trace.inputs[i]
+					elif (i["name"] == '_collateral'):
+						collateral_type = trace.inputs[i]
+					elif (i["name"] == '_reserve'):
+						collateral_source = trace.inputs[i]
+
+
+				# Define the address of the liquidator
+				liquidator = trace.from_address
+
+				# Find a transfer before liquidation with a to_address corresponding to the liquidator
+				for trace in traces[:i]:
+					if (trace.classification == Classification.transfer & trace.to_address == liquidator):
+						profit =
+						#Calculate profit
+
+
+				# Tag liquidation
+				event.append(Strategy(strategy=StategyType.liquidation,
+									  traces=[trace],
+									  protocols=[trace.protocol]))
+
+
+		return result