Add high score alert

src/constants.py

@@ -37,6 +37,9 @@
 # threshold to generate an alert for violating UDP
 UDP_SENSITIVITY_THRESHOLD = 0.005
 
+# threshold to generate an alert for high score
+HIGH_SCORE_THRESHOLD_ETH = 10
+
 # relevant addresses
 SETTLEMENT_CONTRACT_ADDRESS = "0x9008D19f58AAbD9eD0D60971565AA8510560ab41"
 MEV_BLOCKER_KICKBACKS_ADDRESSES = [

src/monitoring_tests/high_score_test.py

@@ -0,0 +1,55 @@
+"""
+Checking winner's score and generating an alert if score is very large
+"""
+
+# pylint: disable=logging-fstring-interpolation
+
+from typing import Any
+from src.monitoring_tests.base_test import BaseTest
+from src.apis.orderbookapi import OrderbookAPI
+from src.constants import HIGH_SCORE_THRESHOLD_ETH
+
+
+class HighScoreTest(BaseTest):
+    """
+    This test checks how large is the winning score and raises an alert if score
+    is above certain threshold
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.orderbook_api = OrderbookAPI()
+
+    def compute_winning_score(self, competition_data: dict[str, Any]) -> bool:
+        """
+        This function simply returns the winning score.
+        """
+        solution = competition_data["solutions"][-1]
+        score = int(solution["score"]) / 10**18
+        log_output = "\t".join(
+            [
+                "Large score test:",
+                f"Tx Hash: {competition_data['transactionHash']}",
+                f"Winning Solver: {solution['solver']}",
+                f"Score in ETH: {score}",
+            ]
+        )
+        if score > HIGH_SCORE_THRESHOLD_ETH:
+            self.alert(log_output)
+        return True
+
+    def run(self, tx_hash: str) -> bool:
+        """
+        Wrapper function for the whole test. Checks if solver competition data is retrievable
+        and then checks how large the winning score is.
+        """
+
+        solver_competition_data = self.orderbook_api.get_solver_competition_data(
+            tx_hash
+        )
+        if solver_competition_data is None:
+            return False
+
+        success = self.compute_winning_score(solver_competition_data)
+
+        return success

tests/e2e/high_score_test.py

@@ -0,0 +1,24 @@
+"""
+Tests for large score test.
+"""
+
+import unittest
+from src.monitoring_tests.high_score_test import (
+    HighScoreTest,
+)
+
+
+class TestHighScore(unittest.TestCase):
+    def test_high_score(self) -> None:
+        high_score_test = HighScoreTest()
+        # large score tx
+        tx_hash = "0x5eef22d04a2f30e62df76614decf43e1cc92ab957285a687f182a0191d85d15a"
+        self.assertTrue(high_score_test.run(tx_hash))
+
+        # small score tx
+        tx_hash = "0x37e7dbc2f3df5f31f017634d07855933c6d82f4fda033daebd4377987d2b5ae9"
+        self.assertTrue(high_score_test.run(tx_hash))
+
+
+if __name__ == "__main__":
+    unittest.main()