Bug fix + improvement (#46)

* Bump version

* Update calendar for 2023

* Bugfix + refactor

* add a little comment

* Make 25 constant var

* Fix tests

Author: ali-behjati

pyproject.toml

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

pyth_observer/__init__.py

@@ -100,14 +100,20 @@ async def run(self):
                     if not price_account.aggregate_price_info:
                         raise RuntimeError("Aggregate price info is missing")
 
+                    # When min_publishers is high it means that the price is not production-ready
+                    # yet and it is still being tested. We need no alerting for these prices.
+                    if price_account.min_publishers >= 10:
+                        continue
+
                     states.append(
                         PriceFeedState(
                             symbol=product.attrs["symbol"],
                             asset_type=product.attrs["asset_type"],
                             public_key=price_account.key,
                             status=price_account.aggregate_price_status,
-                            slot_aggregate_attempted=price_account.valid_slot,
-                            slot_aggregate=price_account.aggregate_price_info.pub_slot,
+                            # this is the solana block slot when price account was fetched
+                            latest_block_slot=price_account.slot,
+                            latest_trading_slot=price_account.last_slot,
                             price_aggregate=price_account.aggregate_price_info.price,
                             confidence_interval_aggregate=price_account.aggregate_price_info.confidence_interval,
                             coingecko_price=coingecko_prices.get(product.attrs["base"]),
@@ -119,17 +125,25 @@ async def run(self):
                     )
 
                     for component in price_account.price_components:
+                        publisher_name = (
+                            self.publishers.get(component.publisher_key.key, "")
+                            + f" ({component.publisher_key.key})"
+                        ).strip()
                         states.append(
                             PublisherState(
+                                publisher_name=publisher_name,
                                 symbol=product.attrs["symbol"],
                                 public_key=component.publisher_key,
                                 confidence_interval=component.latest_price_info.confidence_interval,
                                 confidence_interval_aggregate=component.last_aggregate_price_info.confidence_interval,
                                 price=component.latest_price_info.price,
                                 price_aggregate=price_account.aggregate_price_info.price,
                                 slot=component.latest_price_info.pub_slot,
-                                slot_aggregate=component.last_aggregate_price_info.pub_slot,
+                                aggregate_slot=price_account.last_slot,
+                                # this is the solana block slot when price account was fetched
+                                latest_block_slot=price_account.slot,
                                 status=component.latest_price_info.price_status,
+                                aggregate_status=price_account.aggregate_price_status,
                             )
                         )
 

pyth_observer/calendar.py

@@ -9,18 +9,24 @@
 EQUITY_EARLY_CLOSE = datetime.time(13, 0, 0, tzinfo=TZ)
 
 # EQUITY_HOLIDAYS and EQUITY_EARLY_HOLIDAYS will need to be updated each year
+# From https://www.nyse.com/markets/hours-calendars
 EQUITY_HOLIDAYS = [
-    datetime.datetime(2022, 1, 17, tzinfo=TZ).date(),
-    datetime.datetime(2022, 2, 21, tzinfo=TZ).date(),
-    datetime.datetime(2022, 4, 15, tzinfo=TZ).date(),
-    datetime.datetime(2022, 5, 30, tzinfo=TZ).date(),
-    datetime.datetime(2022, 6, 20, tzinfo=TZ).date(),
-    datetime.datetime(2022, 7, 4, tzinfo=TZ).date(),
-    datetime.datetime(2022, 9, 5, tzinfo=TZ).date(),
-    datetime.datetime(2022, 11, 24, tzinfo=TZ).date(),
-    datetime.datetime(2022, 12, 26, tzinfo=TZ).date(),
+    datetime.datetime(2023, 1, 2, tzinfo=TZ).date(),
+    datetime.datetime(2023, 1, 16, tzinfo=TZ).date(),
+    datetime.datetime(2023, 2, 20, tzinfo=TZ).date(),
+    datetime.datetime(2023, 4, 7, tzinfo=TZ).date(),
+    datetime.datetime(2023, 5, 29, tzinfo=TZ).date(),
+    datetime.datetime(2023, 6, 19, tzinfo=TZ).date(),
+    datetime.datetime(2023, 7, 4, tzinfo=TZ).date(),
+    datetime.datetime(2022, 9, 4, tzinfo=TZ).date(),
+    datetime.datetime(2023, 11, 23, tzinfo=TZ).date(),
+    datetime.datetime(2023, 12, 25, tzinfo=TZ).date(),
+]
+
+EQUITY_EARLY_HOLIDAYS = [
+    datetime.datetime(2023, 7, 3, tzinfo=TZ).date(),
+    datetime.datetime(2023, 11, 24, tzinfo=TZ).date(),
 ]
-EQUITY_EARLY_HOLIDAYS = [datetime.datetime(2022, 11, 25, tzinfo=TZ).date()]
 
 
 class HolidayCalendar:

pyth_observer/check/price_feed.py

@@ -19,8 +19,8 @@ class PriceFeedState:
     asset_type: str
     public_key: SolanaPublicKey
     status: PythPriceStatus
-    slot_aggregate_attempted: int
-    slot_aggregate: int
+    latest_block_slot: int
+    latest_trading_slot: int
     price_aggregate: float
     confidence_interval_aggregate: float
     coingecko_price: Optional[float]
@@ -46,10 +46,11 @@ def error_message(self) -> str:
         ...
 
 
-class PriceFeedAggregateCheck(PriceFeedCheck):
+class PriceFeedOfflineCheck(PriceFeedCheck):
     def __init__(self, state: PriceFeedState, config: PriceFeedCheckConfig):
         self.__state = state
         self.__max_slot_distance: int = int(config["max_slot_distance"])
+        self.__abandoned_slot_distance: int = int(config["abandoned_slot_distance"])
 
     def state(self) -> PriceFeedState:
         return self.__state
@@ -64,28 +65,30 @@ def run(self) -> bool:
         if not is_market_open:
             return True
 
-        # Skip if not trading
-        if self.__state.status != PythPriceStatus.TRADING:
-            return True
-
         distance = abs(
-            self.__state.slot_aggregate_attempted - self.__state.slot_aggregate
+            self.__state.latest_block_slot - self.__state.latest_trading_slot
         )
 
         # Pass if distance is less than max slot distance
         if distance < self.__max_slot_distance:
             return True
 
+        # Pass if price has been stale for a long time
+        if distance > self.__abandoned_slot_distance:
+            return True
+
         # Fail
         return False
 
     def error_message(self) -> str:
+        distance = self.__state.latest_block_slot - self.__state.latest_trading_slot
         return dedent(
             f"""
-            {self.__state.symbol} is offline.
+            {self.__state.symbol} is offline (either non-trading/stale).
+            It is not updated for {distance} slots.
 
-            Valid aggregate slot: {self.__state.slot_aggregate}
-            Attempted aggregate slot: {self.__state.slot_aggregate_attempted}
+            Latest trading slot: {self.__state.latest_trading_slot}
+            Block slot: {self.__state.latest_block_slot}
             """
         ).strip()
 
@@ -174,6 +177,10 @@ def state(self) -> PriceFeedState:
         return self.__state
 
     def run(self) -> bool:
+        # Skip if not trading
+        if self.__state.status != PythPriceStatus.TRADING:
+            return True
+
         # Skip if publish time is zero
         if not self.__state.crosschain_price["publish_time"]:
             return True
@@ -264,5 +271,5 @@ def error_message(self) -> str:
     PriceFeedCoinGeckoCheck,
     PriceFeedCrossChainDeviationCheck,
     PriceFeedCrossChainOnlineCheck,
-    PriceFeedAggregateCheck,
+    PriceFeedOfflineCheck,
 ]

pyth_observer/check/publisher.py

@@ -5,14 +5,19 @@
 from pythclient.pythaccounts import PythPriceStatus
 from pythclient.solana import SolanaPublicKey
 
+PUBLISHER_EXCLUSION_DISTANCE = 25
+
 
 @dataclass
 class PublisherState:
+    publisher_name: str
     symbol: str
     public_key: SolanaPublicKey
     status: PythPriceStatus
+    aggregate_status: PythPriceStatus
     slot: int
-    slot_aggregate: int
+    aggregate_slot: int
+    latest_block_slot: int
     price: float
     price_aggregate: float
     confidence_interval: float
@@ -33,11 +38,11 @@ def state(self) -> PublisherState:
     def run(self) -> bool:
         ...
 
-    def error_message(self, publishers: Dict[str, str]) -> str:
+    def error_message(self) -> str:
         ...
 
 
-class PublisherAggregateCheck(PublisherCheck):
+class PublisherWithinAggregateConfidenceCheck(PublisherCheck):
     def __init__(self, state: PublisherState, config: PublisherCheckConfig):
         self.__state = state
         self.__max_interval_distance: int = int(config["max_interval_distance"])
@@ -46,10 +51,23 @@ def state(self) -> PublisherState:
         return self.__state
 
     def run(self) -> bool:
+        # Skip if not trading
+        if self.__state.status != PythPriceStatus.TRADING:
+            return True
+
+        # Skip if aggregate is not trading
+        if self.__state.aggregate_status != PythPriceStatus.TRADING:
+            return True
+
         # Skip if confidence interval is zero
         if self.__state.confidence_interval == 0:
             return True
 
+        # Pass if publisher slot is far from aggregate slot
+        distance = abs(self.__state.slot - self.__state.aggregate_slot)
+        if distance > PUBLISHER_EXCLUSION_DISTANCE:
+            return True
+
         diff = self.__state.price - self.__state.price_aggregate
         intervals_away = abs(diff / self.__state.confidence_interval_aggregate)
 
@@ -60,11 +78,16 @@ def run(self) -> bool:
         # Fail
         return False
 
-    def error_message(self, publishers) -> str:
+    def error_message(self) -> str:
+        diff = self.__state.price - self.__state.price_aggregate
+        intervals_away = abs(diff / self.__state.confidence_interval_aggregate)
+
         return dedent(
             f"""
-            {publishers[self.__state.public_key.key]} price is too far from aggregate.
+            {self.__state.publisher_name} price not within aggregate confidence.
+            It is {intervals_away} times away from confidence.
 
+            Symbol: {self.__state.symbol}
             Publisher price: {self.__state.price} ± {self.__state.confidence_interval}
             Aggregate price: {self.__state.price_aggregate} ± {self.__state.confidence_interval_aggregate}
             """
@@ -81,7 +104,12 @@ def state(self) -> PublisherState:
 
     def run(self) -> bool:
         # Skip if not trading
-        if not self.__state.status == PythPriceStatus.TRADING:
+        if self.__state.status != PythPriceStatus.TRADING:
+            return True
+
+        # Pass if publisher slot is far from aggregate slot
+        distance = abs(self.__state.slot - self.__state.aggregate_slot)
+        if distance > PUBLISHER_EXCLUSION_DISTANCE:
             return True
 
         # Pass if confidence interval is greater than min_confidence_interval
@@ -91,11 +119,12 @@ def run(self) -> bool:
         # Fail
         return False
 
-    def error_message(self, publishers) -> str:
+    def error_message(self) -> str:
         return dedent(
             f"""
-            {publishers[self.__state.public_key.key]} confidence interval is too tight.
+            {self.__state.publisher_name} confidence interval is too tight.
 
+            Symbol: {self.__state.symbol}
             Price: {self.__state.price}
             Confidence interval: {self.__state.confidence_interval}
             """
@@ -106,27 +135,34 @@ class PublisherOfflineCheck(PublisherCheck):
     def __init__(self, state: PublisherState, config: PublisherCheckConfig):
         self.__state = state
         self.__max_slot_distance: int = int(config["max_slot_distance"])
+        self.__abandoned_slot_distance: int = int(config["abandoned_slot_distance"])
 
     def state(self) -> PublisherState:
         return self.__state
 
     def run(self) -> bool:
-        distance = abs(self.__state.slot - self.__state.slot_aggregate)
+        distance = self.__state.latest_block_slot - self.__state.slot
 
         # Pass if publisher slot is not too far from aggregate slot
-        if distance < 25:
+        if distance < self.__max_slot_distance:
+            return True
+
+        # Pass if publisher has been inactive for a long time
+        if distance > self.__abandoned_slot_distance:
             return True
 
         # Fail
-        return True
+        return False
 
-    def error_message(self, publishers) -> str:
+    def error_message(self) -> str:
+        distance = self.__state.latest_block_slot - self.__state.slot
         return dedent(
             f"""
-            {publishers[self.__state.public_key.key]} hasn't published recently.
+            {self.__state.publisher_name} hasn't published recently for {distance} slots.
 
+            Symbol: {self.__state.symbol}
             Publisher slot: {self.__state.slot}
-            Aggregate slot: {self.__state.slot_aggregate}
+            Aggregate slot: {self.__state.aggregate_slot}
             """
         ).strip()
 
@@ -141,47 +177,51 @@ def state(self) -> PublisherState:
         return self.__state
 
     def run(self) -> bool:
-        price_diff = abs(self.__state.price - self.__state.price_aggregate)
-        slot_diff = abs(self.__state.slot - self.__state.slot_aggregate)
+        # Skip if aggregate status is not trading
+        if self.__state.aggregate_status != PythPriceStatus.TRADING:
+            return True
 
         # Skip if not trading
         if self.__state.status != PythPriceStatus.TRADING:
             return True
 
         # Skip if publisher is too far behind
+        slot_diff = abs(self.__state.slot - self.__state.aggregate_slot)
         if slot_diff > self.__max_slot_distance:
             return True
 
-        # Skip if no aggregate
-        if self.__state.price_aggregate == 0:
+        # Skip if published price is zero
+        if self.__state.price == 0:
             return True
 
-        distance = (price_diff / self.__state.price_aggregate) * 100
+        price_diff = abs(self.__state.price - self.__state.price_aggregate)
+        deviation = (price_diff / self.__state.price_aggregate) * 100
 
         # Pass if deviation is less than max distance
-        if distance <= self.__max_aggregate_distance:
+        if deviation <= self.__max_aggregate_distance:
             return True
 
         # Fail
         return False
 
-    def error_message(self, publishers) -> str:
+    def error_message(self) -> str:
         price_diff = abs(self.__state.price - self.__state.price_aggregate)
-        distance = (price_diff / self.__state.price_aggregate) * 100
+        deviation = (price_diff / self.__state.price_aggregate) * 100
 
         return dedent(
             f"""
-            {publishers[self.__state.public_key.key]} price is too far from aggregate.
+            {self.__state.publisher_name} price is too far from aggregate price.
 
+            Symbol: {self.__state.symbol}
             Publisher price: {self.__state.price}
             Aggregate price: {self.__state.price_aggregate}
-            Distance: {distance}%
+            Deviation: {deviation}%
             """
         ).strip()
 
 
 PUBLISHER_CHECKS = [
-    PublisherAggregateCheck,
+    PublisherWithinAggregateConfidenceCheck,
     PublisherConfidenceIntervalCheck,
     PublisherOfflineCheck,
     PublisherPriceCheck,