diff --git a/src/constants.py b/src/constants.py index b0795ee..91be801 100644 --- a/src/constants.py +++ b/src/constants.py @@ -40,6 +40,9 @@ # threshold to generate an alert for high score HIGH_SCORE_THRESHOLD_ETH = 10 +# threshold to generate an alert for price discrepancy test +UCP_VS_NATIVE_SENSITIVITY_THRESHOLD = 0.5 + # relevant addresses SETTLEMENT_CONTRACT_ADDRESS = "0x9008D19f58AAbD9eD0D60971565AA8510560ab41" MEV_BLOCKER_KICKBACKS_ADDRESSES = [ diff --git a/src/daemon.py b/src/daemon.py index 9c638c0..868913c 100644 --- a/src/daemon.py +++ b/src/daemon.py @@ -20,6 +20,7 @@ from src.monitoring_tests.high_score_test import ( HighScoreTest, ) +from src.monitoring_tests.price_sensitivity_test import PriceSensitivityTest from src.constants import SLEEP_TIME_IN_SEC, CHAIN_ID_TO_NAME @@ -36,6 +37,7 @@ def main() -> None: tests = [ SolverCompetitionSurplusTest(orderbook_api), HighScoreTest(orderbook_api), + PriceSensitivityTest(orderbook_api), ] # special case for mainnet as MEV Blocker only exists on mainnet if chain_name == "mainnet": diff --git a/src/monitoring_tests/price_sensitivity_test.py b/src/monitoring_tests/price_sensitivity_test.py new file mode 100644 index 0000000..bf748c4 --- /dev/null +++ b/src/monitoring_tests/price_sensitivity_test.py @@ -0,0 +1,83 @@ +""" +Checks whether native prices are far from UCP for a trade +""" + +# pylint: disable=duplicate-code +# pylint: disable=too-many-locals +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 ( + UCP_VS_NATIVE_SENSITIVITY_THRESHOLD, +) + + +class PriceSensitivityTest(BaseTest): + """ + This test checks whether the exchange rate implied by native prices + is far from exchange rate implied by UCP + """ + + def __init__(self, orderbook_api: OrderbookAPI) -> None: + super().__init__() + self.orderbook_api = orderbook_api + + def check_prices(self, competition_data: dict[str, Any]) -> bool: + """ + This function checks whether native prices are far from ucp + """ + winning_solution = competition_data["solutions"][-1] + trades_dict = self.orderbook_api.get_uid_trades(winning_solution) + if trades_dict is None: + return False + + ucp = { + token.lower(): int(price) + for token, price in winning_solution["clearingPrices"].items() + } + + native_prices = { + token.lower(): int(price) + for token, price in competition_data["auction"]["prices"].items() + } + + for uid, trade in trades_dict.items(): + sell_token = trade.data.sell_token.lower() + buy_token = trade.data.buy_token.lower() + ucp_rate = Fraction(ucp[sell_token], ucp[buy_token]) + native_price_rate = Fraction( + native_prices[sell_token], native_prices[buy_token] + ) + + max_rate = max(ucp_rate, native_price_rate) + min_rate = min(ucp_rate, native_price_rate) + + if max_rate > (1 + UCP_VS_NATIVE_SENSITIVITY_THRESHOLD) * min_rate: + log_output = "\t".join( + [ + "Price sensitivity test:", + f"Tx Hash: {competition_data['transactionHashes'][0]}", + f"Winning Solver: {winning_solution['solver']}", + f"Trade: {uid}", + f"Gap: {float(max_rate / min_rate)}", + ] + ) + self.alert(log_output) + + return True + + def run(self, tx_hash: str) -> bool: + """ + Wrapper function for the whole test. Checks if violation is more than + UCP_VS_NATIVE_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_prices(solver_competition_data) + + return success diff --git a/tests/e2e/price_sensitivity_test.py b/tests/e2e/price_sensitivity_test.py new file mode 100644 index 0000000..289430b --- /dev/null +++ b/tests/e2e/price_sensitivity_test.py @@ -0,0 +1,22 @@ +""" +Tests for surplus test. +""" + +import unittest +from src.apis.orderbookapi import OrderbookAPI +from src.monitoring_tests.price_sensitivity_test import ( + PriceSensitivityTest, +) + + +class TestPrices(unittest.TestCase): + def test_prices(self) -> None: + orderbook_api = OrderbookAPI("mainnet") + price_sensitivity_test = PriceSensitivityTest(orderbook_api) + # new competition format: no alert or info + tx_hash = "0x524b92cfc81e9b590a701bf157ca1b0f21b05d7c37ebb39f332cd215c8db1046" + self.assertTrue(price_sensitivity_test.run(tx_hash)) + + +if __name__ == "__main__": + unittest.main()