[CIP-38] Add Uniform Directional Clearing Prices test

src/constants.py

@@ -31,6 +31,9 @@
 # threshold parameter to generate an alert when receiving kickbacks
 KICKBACKS_ALERT_THRESHOLD = 0.1
 
+# threshold to generate an alert for violating UDP
+UDP_SENSITIVITY_THRESHOLD = 0.005
+
 # relevant addresses
 SETTLEMENT_CONTRACT_ADDRESS = "0x9008D19f58AAbD9eD0D60971565AA8510560ab41"
 MEV_BLOCKER_KICKBACKS_ADDRESS = "0xCe91228789B57DEb45e66Ca10Ff648385fE7093b"

src/daemon.py

@@ -27,6 +27,9 @@
 from src.monitoring_tests.cost_coverage_zero_signed_fee import (
     CostCoverageForZeroSignedFee,
 )
+from src.monitoring_tests.uniform_directed_prices_test import (
+    UniformDirectedPricesTest,
+)
 from src.constants import SLEEP_TIME_IN_SEC
 
 
@@ -44,6 +47,7 @@ def main() -> None:
         BuffersMonitoringTest(),
         CombinatorialAuctionSurplusTest(),
         CostCoverageForZeroSignedFee(),
+        UniformDirectedPricesTest(),
     ]
 
     start_block: Optional[int] = None

src/monitoring_tests/uniform_directed_prices_test.py

@@ -0,0 +1,83 @@
+"""
+Checks the uniform directed prices constraint that was introduced with CIP-38
+"""
+# pylint: disable=duplicate-code
+from typing import Any
+from fractions import Fraction
+from src.monitoring_tests.base_test import BaseTest
+from src.apis.orderbookapi import OrderbookAPI
+from src.constants import (
+    UDP_SENSITIVITY_THRESHOLD,
+)
+
+
+class UniformDirectedPricesTest(BaseTest):
+    """
+    This test checks whether the Uniform Directed Prices constraint,
+    as introduced in CIP-38, is satisfied.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.orderbook_api = OrderbookAPI()
+
+    def check_udp(self, competition_data: dict[str, Any]) -> bool:
+        """
+        This function checks whether there are multiple orders in the same directed token pair,
+        and if so, checks whether UDP is satisfied.
+        """
+        solution = competition_data["solutions"][-1]
+        trades_dict = self.orderbook_api.get_uid_trades(solution)
+
+        if trades_dict is None:
+            return False
+
+        directional_prices: dict[tuple[str, str], list[Fraction]] = {}
+        for _, trade in trades_dict.items():
+            token_pair = (
+                trade.data.sell_token.lower(),
+                trade.data.buy_token.lower(),
+            )
+            directional_price = Fraction(
+                trade.execution.sell_amount, trade.execution.buy_amount
+            )
+            if token_pair not in directional_prices:
+                directional_prices[token_pair] = []
+            directional_prices[token_pair].append(directional_price)
+
+        for pair, prices_list in directional_prices.items():
+            if len(prices_list) == 1:
+                continue
+            min_rate = min(prices_list)
+            max_rate = max(prices_list)
+
+            log_output = "\t".join(
+                [
+                    "Uniform Directed Prices test:",
+                    f"Tx Hash: {competition_data['transactionHash']}",
+                    f"Winning Solver: {solution['solver']}",
+                    f"Token pair: {pair}",
+                    f"Directional prices: {[float(p) for p in prices_list]}",
+                ]
+            )
+            if max_rate > min_rate * (1 + UDP_SENSITIVITY_THRESHOLD):
+                self.alert(log_output)
+            elif max_rate > min_rate * (1 + UDP_SENSITIVITY_THRESHOLD / 10):
+                self.logger.info(log_output)
+
+        return True
+
+    def run(self, tx_hash: str) -> bool:
+        """
+        Wrapper function for the whole test. Checks if violation is more than
+        UDP_SENSITIVITY_THRESHOLD, in which case it generates an alert.
+        """
+        solver_competition_data = self.orderbook_api.get_solver_competition_data(
+            tx_hash
+        )
+        if solver_competition_data is None:
+            return False
+
+        success = self.check_udp(solver_competition_data)
+
+        return success

tests/e2e/uniform_directional_prices_test.py

@@ -0,0 +1,23 @@
+"""
+Tests for combinatorial auction surplus test.
+"""
+
+import unittest
+from src.monitoring_tests.uniform_directed_prices_test import (
+    UniformDirectedPricesTest,
+)
+
+
+class TestCombinatorialAuctionSurplus(unittest.TestCase):
+    def test_surplus(self) -> None:
+        surplus_test = UniformDirectedPricesTest()
+        # Fair prices
+        # tx_hash = "0x4c702f9a3e4593a16fed03229cb4d449a48eab5fb92030fc8ba596e78fef8d1c"
+        # self.assertTrue(surplus_test.run(tx_hash))
+        # Unfair prices
+        tx_hash = "0x469ac0d0e430c67c94d26ae202e9e6710396a1968b1a6656be002eb4f2b7af65"
+        self.assertTrue(surplus_test.run(tx_hash))
+
+
+if __name__ == "__main__":
+    unittest.main()