add PublisherStalledCheck

pyproject.toml

@@ -4,7 +4,7 @@ ignore_missing_imports = true
 
 [tool.poetry]
 name = "pyth-observer"
-version = "0.2.4"
+version = "0.2.5"
 description = "Alerts and stuff"
 authors = []
 readme = "README.md"

pyth_observer/check/publisher.py

@@ -1,3 +1,4 @@
+import time
 from dataclasses import dataclass
 from typing import Dict, Protocol, runtime_checkable
 
@@ -6,6 +7,8 @@
 
 PUBLISHER_EXCLUSION_DISTANCE = 25
 
+PUBLISHER_CACHE = {}
+
 
 @dataclass
 class PublisherState:
@@ -216,9 +219,48 @@ def ci_adjusted_price_diff(self) -> float:
  return max(price_only_diff - self.__state.confidence_interval, 0)
 
 
+class PublisherStalledCheck(PublisherCheck):
+ def __init__(self, state: PublisherState, config: PublisherCheckConfig):
+ self.__state = state
+ self.__stall_time_limit: int = int(
+ config["stall_time_limit"]
+ ) # Time in seconds
+
+ def state(self) -> PublisherState:
+ return self.__state
+
+ def run(self) -> bool:
+ publisher_key = (self.__state.publisher_name, self.__state.symbol)
+ current_time = time.time()
+ previous_price, last_change_time = PUBLISHER_CACHE.get(
+ publisher_key, (None, None)
+ )
+
+ if previous_price is None or self.__state.price != previous_price:
+ PUBLISHER_CACHE[publisher_key] = (self.__state.price, current_time)
+ return True
+
+ if (current_time - last_change_time) > self.__stall_time_limit:
+ return False
+
+ return True
+
+ def error_message(self) -> dict:
+ return {
+ "msg": f"{self.__state.publisher_name} has been publishing the same price for too long.",
+ "type": "PublisherStalledCheck",
+ "publisher": self.__state.publisher_name,
+ "symbol": self.__state.symbol,
+ "price": self.__state.price,
+ "stall_duration": time.time()
+ - PUBLISHER_CACHE[(self.__state.publisher_name, self.__state.symbol)][1],
+ }
+
+
 PUBLISHER_CHECKS = [
  PublisherWithinAggregateConfidenceCheck,
  PublisherConfidenceIntervalCheck,
  PublisherOfflineCheck,
  PublisherPriceCheck,
+ PublisherStalledCheck,
 ]

sample.config.yaml

@@ -44,6 +44,9 @@ checks:
  enable: true
  max_slot_distance: 25
  max_aggregate_distance: 6
+ PublisherStalledCheck:
+ enable: true
+ stall_time_limit: 60
  # Per-symbol config
  Crypto.MNGO/USD:
  PriceFeedOfflineCheck:

tests/test_checks_publisher.py

@@ -1,7 +1,15 @@
+import time
+from unittest.mock import patch
+
 from pythclient.pythaccounts import PythPriceStatus
 from pythclient.solana import SolanaPublicKey
 
-from pyth_observer.check.publisher import PublisherPriceCheck, PublisherState
+from pyth_observer.check.publisher import (
+ PUBLISHER_CACHE,
+ PublisherPriceCheck,
+ PublisherStalledCheck,
+ PublisherState,
+)
 
 
 def make_state(
@@ -44,3 +52,46 @@ def check_is_ok(
  state1 = make_state(1, 100.0, 2.0, 1, 110.0, 1.0)
  assert check_is_ok(state1, 10, 25)
  assert not check_is_ok(state1, 6, 25)
+
+
+def test_publisher_stalled_check():
+ current_time = time.time()
+
+ def simulate_time_pass(seconds):
+ nonlocal current_time
+ current_time += seconds
+ return current_time
+
+ def setup_check(state, stall_time_limit):
+ check = PublisherStalledCheck(state, {"stall_time_limit": stall_time_limit})
+ PUBLISHER_CACHE[(state.publisher_name, state.symbol)] = (
+ state.price,
+ current_time,
+ )
+ return check
+
+ def run_check(check, seconds, expected):
+ with patch("time.time", new=lambda: simulate_time_pass(seconds)):
+ assert check.run() == expected
+
+ PUBLISHER_CACHE.clear()
+ state_a = make_state(1, 100.0, 2.0, 1, 100.0, 1.0)
+ check_a = setup_check(state_a, 5)
+ run_check(check_a, 5, True) # Should pass as it hits the limit exactly
+
+ PUBLISHER_CACHE.clear()
+ state_b = make_state(1, 100.0, 2.0, 1, 100.0, 1.0)
+ check_b = setup_check(state_b, 5)
+ run_check(check_b, 6, False) # Should fail as it exceeds the limit
+
+ PUBLISHER_CACHE.clear()
+ state_c = make_state(1, 100.0, 2.0, 1, 100.0, 1.0)
+ check_c = setup_check(state_c, 5)
+ run_check(check_c, 2, True) # Initial check should pass
+ state_c.price = 105.0 # Change the price
+ run_check(check_c, 3, True) # Should pass as price changes
+ state_c.price = 100.0 # Change back to original price
+ run_check(check_c, 4, True) # Should pass as price changes
+ run_check(
+ check_c, 8, False
+ ) # Should fail as price stalls for too long after last change