From dc0407317b5c0cb453b094869bd7dc2ac4cd97a3 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 29 Jun 2026 18:28:37 +0000 Subject: [PATCH 01/10] feat(pegs): shared peg-monitoring foundation (#299) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Establish the building blocks consumed by all peg/oracle monitoring layers (L1 market depeg, L2 oracle health, L3 event consumers): - Promote ChainlinkAggregator.json to common-abi/ and repoint protocols/ustb. - Add utils/chainlink.py: batched latestRoundData reader (read_feeds) plus pure, unit-tested helpers — scale_price, is_stale(updated_at, heartbeat, buffer), and round/answeredInRound sanity checks. Generalizes the logic previously inline in protocols/ustb/main.py. - Add utils/pegged_assets.py: PeggedAsset dataclass + PegTarget enum (USD=1, BTC=live BTC/USD via DeFiLlama) as the single registry consumed by L1/L2/L3, with optional chainlink_feed (+heartbeat) and rate_oracle (monotonic) refs. depeg_pct expresses peg deviation, not an absolute floor. - Registry covers USDC, USDT, USDS, USDe, cUSD, iUSD, siUSD, cbBTC, LBTC. Token addresses and Chainlink feeds verified on mainnet; siUSD is a marked placeholder pending address verification. - Unit tests for staleness, round sanity, deviation, and registry helpers. ruff + mypy clean (new files); 42 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../ChainlinkAggregator.json | 0 protocols/ustb/main.py | 2 +- tests/test_chainlink.py | 94 +++++++ tests/test_pegged_assets.py | 101 ++++++++ utils/chainlink.py | 171 +++++++++++++ utils/pegged_assets.py | 240 ++++++++++++++++++ 6 files changed, 607 insertions(+), 1 deletion(-) rename {protocols/ustb/abi => common-abi}/ChainlinkAggregator.json (100%) create mode 100644 tests/test_chainlink.py create mode 100644 tests/test_pegged_assets.py create mode 100644 utils/chainlink.py create mode 100644 utils/pegged_assets.py diff --git a/protocols/ustb/abi/ChainlinkAggregator.json b/common-abi/ChainlinkAggregator.json similarity index 100% rename from protocols/ustb/abi/ChainlinkAggregator.json rename to common-abi/ChainlinkAggregator.json diff --git a/protocols/ustb/main.py b/protocols/ustb/main.py index 40bb0bf7..e068027d 100644 --- a/protocols/ustb/main.py +++ b/protocols/ustb/main.py @@ -48,7 +48,7 @@ # ABIs # --------------------------------------------------------------------------- ABI_ORACLE = load_abi("protocols/ustb/abi/SuperstateOracle.json") -ABI_CHAINLINK = load_abi("protocols/ustb/abi/ChainlinkAggregator.json") +ABI_CHAINLINK = load_abi("common-abi/ChainlinkAggregator.json") ABI_ERC20 = load_abi("common-abi/ERC20.json") USTB_DECIMALS = 6 diff --git a/tests/test_chainlink.py b/tests/test_chainlink.py new file mode 100644 index 00000000..032594c0 --- /dev/null +++ b/tests/test_chainlink.py @@ -0,0 +1,94 @@ +import unittest +from decimal import Decimal + +from utils.chainlink import ( + FeedReading, + RoundData, + is_round_healthy, + is_stale, + round_issues, + scale_price, +) + + +def _round( + round_id: int = 10, + answer: int = 100_000_000, + started_at: int = 1_000, + updated_at: int = 1_000, + answered_in_round: int = 10, +) -> RoundData: + return RoundData(round_id, answer, started_at, updated_at, answered_in_round) + + +class TestScalePrice(unittest.TestCase): + def test_scales_by_decimals(self): + self.assertEqual(scale_price(100_000_000, 8), Decimal("1")) + + def test_scales_fractional(self): + self.assertEqual(scale_price(99_960_043, 8), Decimal("0.99960043")) + + def test_zero_decimals_is_identity(self): + self.assertEqual(scale_price(42, 0), Decimal("42")) + + def test_negative_decimals_raises(self): + with self.assertRaises(ValueError): + scale_price(1, -1) + + +class TestIsStale(unittest.TestCase): + def test_fresh_within_heartbeat(self): + self.assertFalse(is_stale(updated_at=1_000, heartbeat=3_600, now=4_000)) + + def test_stale_past_heartbeat(self): + self.assertTrue(is_stale(updated_at=1_000, heartbeat=3_600, now=5_000)) + + def test_buffer_extends_window(self): + # 4000s elapsed, 3600 heartbeat -> stale, but a 600s buffer keeps it fresh. + self.assertFalse(is_stale(updated_at=1_000, heartbeat=3_600, now=5_000, buffer=600)) + + def test_exactly_at_heartbeat_is_not_stale(self): + self.assertFalse(is_stale(updated_at=1_000, heartbeat=3_600, now=4_600)) + + def test_uninitialised_updated_at_is_stale(self): + self.assertTrue(is_stale(updated_at=0, heartbeat=3_600, now=4_000)) + + +class TestRoundSanity(unittest.TestCase): + def test_healthy_round_has_no_issues(self): + self.assertEqual(round_issues(_round()), []) + self.assertTrue(is_round_healthy(_round())) + + def test_non_positive_answer(self): + issues = round_issues(_round(answer=0)) + self.assertTrue(any("non-positive answer" in i for i in issues)) + self.assertFalse(is_round_healthy(_round(answer=-5))) + + def test_incomplete_round(self): + issues = round_issues(_round(updated_at=0)) + self.assertTrue(any("not complete" in i for i in issues)) + + def test_stale_answered_in_round(self): + issues = round_issues(_round(round_id=10, answered_in_round=9)) + self.assertTrue(any("stale round" in i for i in issues)) + + +class TestRoundData(unittest.TestCase): + def test_from_tuple_decodes_fields(self): + rd = RoundData.from_tuple((10, 99_851_375, 900, 1_000, 10)) + self.assertEqual(rd.round_id, 10) + self.assertEqual(rd.answer, 99_851_375) + self.assertEqual(rd.updated_at, 1_000) + self.assertEqual(rd.answered_in_round, 10) + + def test_from_tuple_wrong_length_raises(self): + with self.assertRaises(ValueError): + RoundData.from_tuple((1, 2, 3)) + + def test_feed_reading_price_uses_decimals(self): + reading = FeedReading(address="0xabc", round_data=_round(answer=605_044_986_7456), decimals=8) + self.assertEqual(reading.price, scale_price(605_044_986_7456, 8)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pegged_assets.py b/tests/test_pegged_assets.py new file mode 100644 index 00000000..a15de9ff --- /dev/null +++ b/tests/test_pegged_assets.py @@ -0,0 +1,101 @@ +import unittest +from decimal import Decimal +from unittest.mock import patch + +from utils.pegged_assets import ( + BTC_USD_DEFILLAMA_KEY, + PEGGED_ASSETS, + PEGGED_ASSETS_BY_NAME, + PegTarget, + get_asset, + price_deviation, + resolve_peg_prices, +) + +# Asset set the registry must cover per the issue acceptance criteria. +REQUIRED_ASSETS = {"cbBTC", "LBTC", "siUSD", "cUSD", "USDe", "USDC", "USDT", "USDS"} + + +class TestPriceDeviation(unittest.TestCase): + def test_no_deviation(self): + self.assertEqual(price_deviation(Decimal("1"), Decimal("1")), Decimal("0")) + + def test_positive_deviation(self): + self.assertEqual(price_deviation(Decimal("1.05"), Decimal("1")), Decimal("0.05")) + + def test_negative_deviation(self): + self.assertEqual(price_deviation(Decimal("0.97"), Decimal("1")), Decimal("-0.03")) + + def test_btc_denominated_deviation(self): + # asset at 60,300 vs 60,000 BTC peg -> +0.5% + self.assertEqual(price_deviation(Decimal("60300"), Decimal("60000")), Decimal("0.005")) + + def test_zero_peg_raises(self): + with self.assertRaises(ValueError): + price_deviation(Decimal("1"), Decimal("0")) + + +class TestIsDepegged(unittest.TestCase): + def test_within_tolerance_is_not_depegged(self): + usdc = get_asset("USDC") # depeg_pct = 0.02 + self.assertFalse(usdc.is_depegged(Decimal("0.99"), Decimal("1"))) + + def test_beyond_tolerance_is_depegged(self): + usdc = get_asset("USDC") + self.assertTrue(usdc.is_depegged(Decimal("0.97"), Decimal("1"))) + + def test_at_threshold_is_depegged(self): + usdc = get_asset("USDC") + self.assertTrue(usdc.is_depegged(Decimal("1.02"), Decimal("1"))) + + def test_btc_asset_uses_peg_price(self): + cbbtc = get_asset("cbBTC") # depeg_pct = 0.02, peg = BTC + self.assertFalse(cbbtc.is_depegged(Decimal("60500"), Decimal("60000"))) + self.assertTrue(cbbtc.is_depegged(Decimal("58000"), Decimal("60000"))) + + +class TestResolvePegPrices(unittest.TestCase): + def test_usd_only_does_not_hit_network(self): + with patch("utils.pegged_assets.fetch_prices") as mock_fetch: + prices = resolve_peg_prices({PegTarget.USD}) + mock_fetch.assert_not_called() + self.assertEqual(prices, {PegTarget.USD: Decimal(1)}) + + def test_btc_fetches_live_price(self): + with patch("utils.pegged_assets.fetch_prices", return_value={BTC_USD_DEFILLAMA_KEY: Decimal("60000")}): + prices = resolve_peg_prices({PegTarget.USD, PegTarget.BTC}) + self.assertEqual(prices[PegTarget.USD], Decimal(1)) + self.assertEqual(prices[PegTarget.BTC], Decimal("60000")) + + def test_missing_btc_price_raises(self): + with patch("utils.pegged_assets.fetch_prices", return_value={}): + with self.assertRaises(ValueError): + resolve_peg_prices({PegTarget.BTC}) + + +class TestRegistry(unittest.TestCase): + def test_covers_required_assets(self): + self.assertTrue(REQUIRED_ASSETS.issubset(set(PEGGED_ASSETS_BY_NAME))) + + def test_names_are_unique(self): + names = [a.name for a in PEGGED_ASSETS] + self.assertEqual(len(names), len(set(names))) + + def test_address_parsed_from_defillama_key(self): + self.assertEqual(get_asset("USDC").address, "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48") + + def test_btc_pegged_assets_target_btc(self): + self.assertEqual(get_asset("cbBTC").peg, PegTarget.BTC) + self.assertEqual(get_asset("LBTC").peg, PegTarget.BTC) + + def test_get_asset_unknown_raises(self): + with self.assertRaises(KeyError): + get_asset("NOPE") + + def test_every_asset_has_positive_depeg_tolerance(self): + for asset in PEGGED_ASSETS: + self.assertGreater(asset.depeg_pct, Decimal("0"), asset.name) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/chainlink.py b/utils/chainlink.py new file mode 100644 index 00000000..79b50bf1 --- /dev/null +++ b/utils/chainlink.py @@ -0,0 +1,171 @@ +"""Chainlink aggregator helpers shared across peg / oracle monitors. + +Generalises the inline ``latestRoundData`` handling from ``protocols/ustb/main.py``: +a batched feed reader plus pure helpers for staleness, round sanity checks and +price scaling. Pure helpers take primitive values so they are trivially unit +testable without a chain connection. +""" + +from dataclasses import dataclass +from decimal import Decimal + +from utils.abi import load_abi +from utils.logger import get_logger +from utils.web3_wrapper import Web3Client + +logger = get_logger("chainlink") + +# Shared Chainlink AggregatorV3Interface ABI (latestRoundData + decimals). +CHAINLINK_ABI = load_abi("common-abi/ChainlinkAggregator.json") + + +@dataclass(frozen=True) +class RoundData: + """Decoded Chainlink ``latestRoundData`` tuple.""" + + round_id: int + answer: int + started_at: int + updated_at: int + answered_in_round: int + + @classmethod + def from_tuple(cls, data: tuple | list) -> "RoundData": + """Build a ``RoundData`` from a raw ``latestRoundData`` response. + + Args: + data: The 5-element tuple/list returned by ``latestRoundData``. + + Raises: + ValueError: If ``data`` does not have exactly five elements. + """ + if len(data) != 5: + raise ValueError(f"latestRoundData expects 5 fields, got {len(data)}: {data!r}") + return cls( + round_id=int(data[0]), + answer=int(data[1]), + started_at=int(data[2]), + updated_at=int(data[3]), + answered_in_round=int(data[4]), + ) + + +@dataclass(frozen=True) +class FeedReading: + """A single feed's decoded round data, decimals and scaled price.""" + + address: str + round_data: RoundData + decimals: int + + @property + def price(self) -> Decimal: + """Answer scaled to a human-readable value by the feed's decimals.""" + return scale_price(self.round_data.answer, self.decimals) + + +# --------------------------------------------------------------------------- +# Pure helpers (no chain connection required — unit tested directly) +# --------------------------------------------------------------------------- + + +def scale_price(answer: int, decimals: int) -> Decimal: + """Scale a raw integer answer to a decimal price using the feed decimals. + + Args: + answer: Raw integer answer from the aggregator. + decimals: Number of decimals reported by the feed. + + Returns: + The answer divided by ``10 ** decimals`` as a ``Decimal``. + + Raises: + ValueError: If ``decimals`` is negative. + """ + if decimals < 0: + raise ValueError(f"decimals must be non-negative, got {decimals}") + return Decimal(answer) / (Decimal(10) ** decimals) + + +def is_stale(updated_at: int, heartbeat: int, now: int, buffer: int = 0) -> bool: + """Return ``True`` if a feed has not updated within its heartbeat window. + + Args: + updated_at: ``updatedAt`` timestamp from the latest round (unix seconds). + heartbeat: Expected maximum interval between updates (seconds). + now: Current unix timestamp (seconds). + buffer: Extra grace period added to the heartbeat before flagging + staleness (seconds). Defaults to ``0``. + + Returns: + ``True`` when ``now - updated_at`` exceeds ``heartbeat + buffer``, or when + ``updated_at`` is non-positive (uninitialised / invalid round). + """ + if updated_at <= 0: + return True + return (now - updated_at) > (heartbeat + buffer) + + +def round_issues(round_data: RoundData) -> list[str]: + """Collect Chainlink round sanity-check failures. + + Checks the standard freshness/consistency invariants Chainlink consumers are + expected to enforce: a positive answer, an initialised round, and an + ``answeredInRound`` that is not behind the current ``roundId`` (a stale answer + carried over from an earlier round). + + Args: + round_data: The decoded round data to validate. + + Returns: + A list of human-readable problem descriptions; empty when healthy. + """ + issues: list[str] = [] + if round_data.answer <= 0: + issues.append(f"non-positive answer ({round_data.answer})") + if round_data.updated_at <= 0: + issues.append("round not complete (updatedAt is 0)") + if round_data.answered_in_round < round_data.round_id: + issues.append(f"stale round (answeredInRound {round_data.answered_in_round} < roundId {round_data.round_id})") + return issues + + +def is_round_healthy(round_data: RoundData) -> bool: + """Return ``True`` if the round passes all sanity checks in :func:`round_issues`.""" + return not round_issues(round_data) + + +# --------------------------------------------------------------------------- +# Batched on-chain reader +# --------------------------------------------------------------------------- + + +def read_feeds(client: Web3Client, feed_addresses: list[str]) -> dict[str, FeedReading]: + """Read ``latestRoundData`` and ``decimals`` for several feeds in one batch. + + Args: + client: Connected ``Web3Client`` for the target chain. + feed_addresses: Chainlink aggregator addresses to read. + + Returns: + Mapping of feed address to its :class:`FeedReading`, preserving input order. + """ + if not feed_addresses: + return {} + + contracts = [client.get_contract(address, CHAINLINK_ABI) for address in feed_addresses] + + with client.batch_requests() as batch: + for contract in contracts: + batch.add(contract.functions.latestRoundData()) + batch.add(contract.functions.decimals()) + responses = client.execute_batch(batch) + + readings: dict[str, FeedReading] = {} + for index, address in enumerate(feed_addresses): + round_data = RoundData.from_tuple(responses[2 * index]) + decimals = int(responses[2 * index + 1]) + readings[address] = FeedReading(address=address, round_data=round_data, decimals=decimals) + logger.info("Chainlink feed %s: price=%s decimals=%d", address, readings[address].price, decimals) + + return readings diff --git a/utils/pegged_assets.py b/utils/pegged_assets.py new file mode 100644 index 00000000..a5b77424 --- /dev/null +++ b/utils/pegged_assets.py @@ -0,0 +1,240 @@ +"""Single source of truth for pegged-asset peg monitoring. + +This registry is consumed by every layer of peg/oracle monitoring: + +* L1 — market depeg (DeFiLlama price vs ``peg`` target, deviation > ``depeg_pct``) +* L2 — oracle health (``chainlink_feed`` staleness / round sanity, ``rate_oracle`` drift) +* L3 — event consumers + +Peg deviation is expressed relative to a :class:`PegTarget` (``USD`` is the +constant ``1``; ``BTC`` is the live BTC/USD price from DeFiLlama), so a single +entry covers both dollar- and bitcoin-denominated assets. ``depeg_pct`` is a +*deviation* tolerance (fractional distance from the peg), not an absolute floor. + +Addresses and Chainlink feeds were verified on Ethereum mainnet; entries that +could not be verified use :data:`PLACEHOLDER_ADDRESS` and are marked ``TODO``. +""" + +from dataclasses import dataclass +from decimal import Decimal +from enum import Enum + +from utils.defillama import fetch_prices + +# DeFiLlama key for the live BTC/USD reference price (BTC peg target). +BTC_USD_DEFILLAMA_KEY = "coingecko:bitcoin" + +# Sentinel for assets/feeds whose address has not yet been verified on-chain. +PLACEHOLDER_ADDRESS = "0x0000000000000000000000000000000000000000" + + +class PegTarget(Enum): + """What an asset is pegged to. + + ``USD`` resolves to the constant ``1``; ``BTC`` resolves to the live BTC/USD + price fetched from DeFiLlama. + """ + + USD = "USD" + BTC = "BTC" + + +@dataclass(frozen=True) +class ChainlinkFeed: + """A Chainlink aggregator backing an asset's price (consumed by L2).""" + + address: str + heartbeat: int # max expected seconds between updates (Chainlink mainnet default) + description: str = "" + + +@dataclass(frozen=True) +class RateOracle: + """A continuous / exchange-rate oracle backing a yield-bearing asset (L2/L3). + + Args: + address: Oracle contract address. + monotonic: Whether the rate is expected to be non-decreasing; a decrease + is a loss signal worth alerting on. + function: View function returning the rate. Defaults to ``"rate"``. + precision: Fixed-point precision of the returned rate. Defaults to ``1e18``. + """ + + address: str + monotonic: bool = True + function: str = "rate" + precision: int = 10**18 + description: str = "" + + +@dataclass(frozen=True) +class PeggedAsset: + """A monitored pegged asset and everything the peg layers need to check it.""" + + name: str + defillama_key: str # "chain:address" + channel: str # Telegram routing channel + peg: PegTarget + depeg_pct: Decimal # deviation tolerance from the peg (e.g. Decimal("0.02") = 2%) + chainlink_feed: ChainlinkFeed | None = None + rate_oracle: RateOracle | None = None + + @property + def address(self) -> str: + """Token address parsed from ``defillama_key`` ("chain:address").""" + return self.defillama_key.split(":", 1)[1] + + def deviation(self, price: Decimal, peg_price: Decimal) -> Decimal: + """Signed fractional deviation of ``price`` from ``peg_price``.""" + return price_deviation(price, peg_price) + + def is_depegged(self, price: Decimal, peg_price: Decimal) -> bool: + """Return ``True`` if ``price`` deviates from ``peg_price`` beyond ``depeg_pct``.""" + return abs(self.deviation(price, peg_price)) >= self.depeg_pct + + +# --------------------------------------------------------------------------- +# Pure helpers +# --------------------------------------------------------------------------- + + +def price_deviation(price: Decimal, peg_price: Decimal) -> Decimal: + """Signed fractional deviation of ``price`` from ``peg_price``. + + Args: + price: Observed asset price. + peg_price: Reference peg price. + + Returns: + ``(price - peg_price) / peg_price``. + + Raises: + ValueError: If ``peg_price`` is zero (deviation is undefined). + """ + if peg_price == 0: + raise ValueError("peg_price must be non-zero to compute deviation") + return (price - peg_price) / peg_price + + +def resolve_peg_prices(pegs: set[PegTarget]) -> dict[PegTarget, Decimal]: + """Resolve current prices for a set of peg targets. + + ``USD`` is the constant ``1`` and never hits the network; ``BTC`` is fetched + once from DeFiLlama only when present in ``pegs``. + + Args: + pegs: The distinct peg targets to resolve. + + Returns: + Mapping of each requested :class:`PegTarget` to its current price. + + Raises: + ValueError: If the BTC/USD price is requested but unavailable. + """ + prices: dict[PegTarget, Decimal] = {} + if PegTarget.USD in pegs: + prices[PegTarget.USD] = Decimal(1) + if PegTarget.BTC in pegs: + fetched = fetch_prices([BTC_USD_DEFILLAMA_KEY]) + btc_price = fetched.get(BTC_USD_DEFILLAMA_KEY) + if btc_price is None: + raise ValueError(f"BTC/USD price unavailable from DeFiLlama key {BTC_USD_DEFILLAMA_KEY}") + prices[PegTarget.BTC] = btc_price + return prices + + +def get_asset(name: str) -> PeggedAsset: + """Look up a registered asset by name. + + Raises: + KeyError: If no asset with ``name`` is registered. + """ + return PEGGED_ASSETS_BY_NAME[name] + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +# Chainlink mainnet stable feeds report 8 decimals with a 24h heartbeat unless +# noted; confirm per feed before tightening staleness thresholds in L2. +_STABLE_HEARTBEAT = 86_400 # 24h + +PEGGED_ASSETS: list[PeggedAsset] = [ + # --- USD-pegged blue chips ------------------------------------------------ + PeggedAsset( + name="USDC", + defillama_key="ethereum:0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", + channel="pegs", + peg=PegTarget.USD, + depeg_pct=Decimal("0.02"), + chainlink_feed=ChainlinkFeed("0x8fFfFfd4AfB6115b954Bd326cbe7B4BA576818f6", _STABLE_HEARTBEAT, "USDC/USD"), + ), + PeggedAsset( + name="USDT", + defillama_key="ethereum:0xdAC17F958D2ee523a2206206994597C13D831ec7", + channel="pegs", + peg=PegTarget.USD, + depeg_pct=Decimal("0.02"), + chainlink_feed=ChainlinkFeed("0x3E7d1eAB13ad0104d2750B8863b489D65364e32D", _STABLE_HEARTBEAT, "USDT/USD"), + ), + PeggedAsset( + name="USDS", + defillama_key="ethereum:0xdC035D45d973E3EC169d2276DDab16f1e407384F", + channel="pegs", + peg=PegTarget.USD, + depeg_pct=Decimal("0.02"), + chainlink_feed=ChainlinkFeed("0xfF30586cD0F29eD462364C7e81375FC0C71219b1", _STABLE_HEARTBEAT, "USDS/USD"), + ), + # --- USD-pegged protocol stables ------------------------------------------ + PeggedAsset( + name="USDe", + defillama_key="ethereum:0x4c9EDD5852cd905f086C759E8383e09bff1E68B3", + channel="ethena", + peg=PegTarget.USD, + depeg_pct=Decimal("0.03"), + chainlink_feed=ChainlinkFeed("0xa569d910839Ae8865Da8F8e70FfFb0cBA869F961", _STABLE_HEARTBEAT, "USDe/USD"), + ), + PeggedAsset( + name="cUSD", + defillama_key="ethereum:0xcccc62962d17b8914c62d74ffb843d73b2a3cccc", + channel="cap", + peg=PegTarget.USD, + depeg_pct=Decimal("0.05"), # cap cUSD price is more volatile + ), + PeggedAsset( + name="iUSD", + defillama_key="ethereum:0x48f9e38f3070AD8945DFEae3FA70987722E3D89c", + channel="infinifi", + peg=PegTarget.USD, + depeg_pct=Decimal("0.03"), + ), + PeggedAsset( + # TODO: siUSD token address not yet verified on-chain — placeholder. + name="siUSD", + defillama_key=f"ethereum:{PLACEHOLDER_ADDRESS}", + channel="pegs", + peg=PegTarget.USD, + depeg_pct=Decimal("0.05"), + ), + # --- BTC-pegged ----------------------------------------------------------- + PeggedAsset( + name="cbBTC", + defillama_key="ethereum:0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf", + channel="pegs", + peg=PegTarget.BTC, + depeg_pct=Decimal("0.02"), + chainlink_feed=ChainlinkFeed("0x2665701293fCbEB223D11A08D826563EDcCE423A", _STABLE_HEARTBEAT, "cbBTC/USD"), + ), + PeggedAsset( + name="LBTC", + defillama_key="ethereum:0x8236a87084f8B84306f72007F36F2618A5634494", + channel="pegs", + peg=PegTarget.BTC, + depeg_pct=Decimal("0.03"), + # LBTC/BTC market-rate feed (8 decimals); price ~1 BTC per LBTC. + chainlink_feed=ChainlinkFeed("0x5c29868C58b6e15e2b962943278969Ab6a7D3212", _STABLE_HEARTBEAT, "LBTC/BTC"), + ), +] + +PEGGED_ASSETS_BY_NAME: dict[str, PeggedAsset] = {asset.name: asset for asset in PEGGED_ASSETS} From 1f8250058b2d0a76e892b8ab649f9f934821a0fd Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 29 Jun 2026 18:58:39 +0000 Subject: [PATCH 02/10] =?UTF-8?q?feat(pegs):=20L2=20oracle=20health=20moni?= =?UTF-8?q?tor=20=E2=80=94=20stables/oracles.py=20(#301)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Layer 2 of peg monitoring: an hourly script that watches the on-chain oracles our lending markets actually liquidate on, driven by the shared PeggedAsset registry. Stacks on the foundation branch (#299 / #305). Per Chainlink-backed asset (USDC, USDT, USDS, USDe, cbBTC, LBTC): - staleness (now - updatedAt > heartbeat + buffer), - round sanity / monotonicity (positive answer, completed round, answeredInRound >= roundId, non-decreasing roundId vs cached run), - deviation from peg (oracle price vs PegTarget), - oracle <-> market divergence (Chainlink vs DeFiLlama) — the liquidation signal. Per-check logic is pure (Alert | None) so forced-stale / forced-divergence / off-peg / round-malfunction cases are unit-tested without a chain. Rate / fundamental oracles: generic monotonicity + delta-vs-cached handler (apyusd approach); a monotonic/capped decrease is CRITICAL (per #196). LBTC Redstone is already covered by a Tenderly alert, so it is documented in TENDERLY_COVERED and not double-polled; gap-analysis notes included. Foundation: add ChainlinkFeed.quote (PegTarget) so BTC-denominated feeds (LBTC/BTC) convert to a USD basis for divergence; LBTC feed marked quote=BTC. Wired into the hourly profile in automation/jobs.yaml (renders; 24 tasks). Validation: ruff + mypy clean (new files); 52 peg tests + 13 automation tests pass; live mainnet smoke run reads all 6 feeds and computes peg/market deviations correctly (LBTC $60,405 = 1.0043 BTC x $60,146). Co-Authored-By: Claude Opus 4.8 (1M context) --- automation/jobs.yaml | 1 + protocols/stables/oracles.py | 341 ++++++++++++++++++++++++++++++++++ tests/test_stables_oracles.py | 180 ++++++++++++++++++ utils/pegged_assets.py | 12 +- 4 files changed, 532 insertions(+), 2 deletions(-) create mode 100644 protocols/stables/oracles.py create mode 100644 tests/test_stables_oracles.py diff --git a/automation/jobs.yaml b/automation/jobs.yaml index aafaff12..892b6b1b 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -37,6 +37,7 @@ profiles: - { name: "usdai", script: protocols/usdai/main.py } - { name: "usdai-large-mints", script: protocols/usdai/large_mints.py } - { name: "stables-dune-large-transfers", script: protocols/stables/dune_large_transfers.py } + - { name: "stables-oracles", script: protocols/stables/oracles.py } - { name: "yearn-alert-large-flows", script: protocols/yearn/alert_large_flows.py } # Cache: tks-trigger-cache.json under $CACHE_DIR (check_stuck_triggers.DEFAULT_CACHE_FILE). - { name: "yearn-check-stuck-triggers", script: protocols/yearn/check_stuck_triggers.py, enabled: false } diff --git a/protocols/stables/oracles.py b/protocols/stables/oracles.py new file mode 100644 index 00000000..f71ba770 --- /dev/null +++ b/protocols/stables/oracles.py @@ -0,0 +1,341 @@ +"""Layer 2 peg monitoring — on-chain oracle health for pegged assets (hourly). + +Where ``protocols/stables/main.py`` watches *market* price (DeFiLlama), this watches +the **on-chain oracles our lending markets actually liquidate on**. Driven by the +shared :data:`PeggedAsset` registry, for every Chainlink-backed asset it checks: + +* **staleness** — ``now − updatedAt > heartbeat + buffer``; +* **round sanity / monotonicity** — positive answer, completed round, + ``answeredInRound ≥ roundId``, and a non-decreasing ``roundId`` vs the cached run; +* **deviation from peg** — oracle price vs the asset's :class:`PegTarget`; +* **oracle ↔ market divergence** — Chainlink vs DeFiLlama, the actual + liquidation-risk signal (markets liquidate on the oracle, not market price). + +For **rate / fundamental oracles** (vault-rate, capped Redstone feeds) it checks +monotonicity + delta-vs-cached (the ``protocols/apyusd/main.py`` approach); any +fundamental-oracle depeg is ``CRITICAL`` (per #196). Fundamental oracles already +covered by a Tenderly alert are listed in :data:`TENDERLY_COVERED` and skipped +here to avoid duplicate alerting. + +Runs hourly via ``automation/jobs.yaml``. +""" + +from dataclasses import dataclass +from decimal import Decimal + +from utils.alert import Alert, AlertSeverity, send_alert +from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file +from utils.chainlink import FeedReading, read_feeds, round_issues +from utils.chains import Chain +from utils.config import Config +from utils.defillama import fetch_prices +from utils.logger import get_logger +from utils.pegged_assets import PEGGED_ASSETS, PeggedAsset, PegTarget, price_deviation, resolve_peg_prices +from utils.web3_wrapper import ChainManager, Web3Client + +PROTOCOL = "pegs" +logger = get_logger("stables-oracles") + +# Tunables (env-overridable). +STALENESS_BUFFER = Config.get_env_int("PEG_ORACLE_STALENESS_BUFFER", 600) # 10 min grace on heartbeat +DIVERGENCE_THRESHOLD = Decimal(str(Config.get_env_float("PEG_ORACLE_DIVERGENCE_THRESHOLD", 0.01))) # 1% +RATE_DELTA_THRESHOLD = Decimal(str(Config.get_env_float("PEG_ORACLE_RATE_DELTA_THRESHOLD", 0.05))) # 5% + +CACHE_FILE = cache_filename + + +def _round_cache_key(address: str) -> str: + return f"peg_oracle_round_{address.lower()}" + + +def _rate_cache_key(address: str) -> str: + return f"peg_oracle_rate_{address.lower()}" + + +# Fundamental oracles already covered by an existing Tenderly alert — NOT polled +# here to avoid duplicate alerting. See protocols/lrt-pegs/README.md. +# +# Gap analysis (per #196 step 6): the registry currently exposes no *uncovered* +# fundamental oracle. Adding active polling for a new one needs the oracle +# contract address, its read function + precision, and whether it is monotonic +# (capped) — wire it as a ``rate_oracle`` on the relevant ``PeggedAsset``. +TENDERLY_COVERED: dict[str, str] = { + # LBTC Redstone fundamental oracle, upper-capped at 1 (healthy == 1). + "LBTC Redstone (0xb415eAA355D8440ac7eCB602D3fb67ccC1f0bc81)": ( + "https://dashboard.tenderly.co/yearn/sam/alerts/rules/eca272ef-979a-47b3-a7f0-2e67172889bb" + ), +} + + +# --------------------------------------------------------------------------- +# Observation + pure per-check helpers (unit tested without a chain) +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class OracleObservation: + """Everything needed to evaluate one Chainlink-backed asset for one run.""" + + asset: PeggedAsset + reading: FeedReading + peg_price_usd: Decimal # USD price of the asset's peg target + quote_price_usd: Decimal # USD price of the feed's quote unit + now: int + market_price_usd: Decimal | None = None # DeFiLlama; None when unavailable + prev_round_id: int | None = None # cached roundId from the previous run + + @property + def oracle_price_usd(self) -> Decimal: + """Chainlink answer expressed in USD (answer × quote price).""" + return self.reading.price * self.quote_price_usd + + +def check_staleness(obs: OracleObservation, buffer: int = STALENESS_BUFFER) -> Alert | None: + """Alert (HIGH) if the feed has not updated within heartbeat + buffer.""" + feed = obs.asset.chainlink_feed + assert feed is not None + updated_at = obs.reading.round_data.updated_at + age = obs.now - updated_at + if updated_at <= 0 or age > feed.heartbeat + buffer: + return Alert( + AlertSeverity.HIGH, + f"*{obs.asset.name} oracle stale* ({feed.description})\n" + f"Age: {age}s — heartbeat {feed.heartbeat}s + buffer {buffer}s\n" + f"updatedAt: {updated_at}\n" + f"Feed: `{feed.address}`", + PROTOCOL, + channel=obs.asset.channel, + ) + return None + + +def check_round_health(obs: OracleObservation) -> Alert | None: + """Alert if round sanity checks fail or ``roundId`` moved backwards. + + A non-positive answer, an incomplete round, or a backwards ``roundId`` is a + feed malfunction (``CRITICAL``); a stale ``answeredInRound`` is ``HIGH``. + """ + feed = obs.asset.chainlink_feed + assert feed is not None + round_data = obs.reading.round_data + issues = round_issues(round_data) + + if obs.prev_round_id is not None and round_data.round_id < obs.prev_round_id: + issues.append(f"roundId went backwards ({obs.prev_round_id} -> {round_data.round_id})") + + if not issues: + return None + + # Anything beyond a lagging answeredInRound means the feed is broken. + critical = any("answeredInRound" not in issue for issue in issues) + severity = AlertSeverity.CRITICAL if critical else AlertSeverity.HIGH + return Alert( + severity, + f"*{obs.asset.name} oracle round unhealthy* ({feed.description})\n" + + "\n".join(f"- {issue}" for issue in issues) + + f"\nFeed: `{feed.address}`", + PROTOCOL, + channel=obs.asset.channel, + ) + + +def check_peg_deviation(obs: OracleObservation) -> Alert | None: + """Alert (HIGH) if the oracle price deviates from the peg beyond ``depeg_pct``.""" + if obs.peg_price_usd <= 0: + return None + if not obs.asset.is_depegged(obs.oracle_price_usd, obs.peg_price_usd): + return None + dev = obs.asset.deviation(obs.oracle_price_usd, obs.peg_price_usd) + return Alert( + AlertSeverity.HIGH, + f"*{obs.asset.name} oracle off peg* ({obs.asset.peg.value})\n" + f"Oracle: ${obs.oracle_price_usd:.6f}\n" + f"Peg: ${obs.peg_price_usd:.6f}\n" + f"Deviation: {dev:+.2%} (tolerance {obs.asset.depeg_pct:.2%})", + PROTOCOL, + channel=obs.asset.channel, + ) + + +def check_market_divergence(obs: OracleObservation, threshold: Decimal = DIVERGENCE_THRESHOLD) -> Alert | None: + """Alert (HIGH) if the oracle and DeFiLlama market price diverge beyond ``threshold``.""" + if obs.market_price_usd is None or obs.market_price_usd <= 0: + return None + dev = price_deviation(obs.oracle_price_usd, obs.market_price_usd) + if abs(dev) < threshold: + return None + return Alert( + AlertSeverity.HIGH, + f"*{obs.asset.name} oracle ↔ market divergence*\n" + f"Oracle: ${obs.oracle_price_usd:.6f}\n" + f"Market (DeFiLlama): ${obs.market_price_usd:.6f}\n" + f"Divergence: {dev:+.2%} (threshold {threshold:.2%})", + PROTOCOL, + channel=obs.asset.channel, + ) + + +def evaluate_chainlink_asset( + obs: OracleObservation, + *, + buffer: int = STALENESS_BUFFER, + divergence_threshold: Decimal = DIVERGENCE_THRESHOLD, +) -> list[Alert]: + """Run all Chainlink-asset checks, returning the alerts that fired.""" + candidates = [ + check_staleness(obs, buffer), + check_round_health(obs), + check_peg_deviation(obs), + check_market_divergence(obs, divergence_threshold), + ] + return [alert for alert in candidates if alert is not None] + + +def check_rate_oracle( + asset: PeggedAsset, + current_rate: int, + prev_rate: int | None, + threshold: Decimal = RATE_DELTA_THRESHOLD, +) -> list[Alert]: + """Monotonicity + delta-vs-cached checks for a fundamental / rate oracle. + + A decrease in a monotonic (capped) oracle is a fundamental depeg + (``CRITICAL``); any delta beyond ``threshold`` is ``HIGH``. + """ + oracle = asset.rate_oracle + assert oracle is not None + if prev_rate is None or prev_rate <= 0: + return [] + + alerts: list[Alert] = [] + delta = Decimal(current_rate - prev_rate) / Decimal(prev_rate) + + if oracle.monotonic and current_rate < prev_rate: + alerts.append( + Alert( + AlertSeverity.CRITICAL, + f"*{asset.name} fundamental oracle DECREASED* (monotonic/capped)\n" + f"Previous: {prev_rate}\nCurrent: {current_rate}\nDelta: {delta:+.4%}\n" + f"Oracle: `{oracle.address}`", + PROTOCOL, + channel=asset.channel, + ) + ) + elif abs(delta) >= threshold: + alerts.append( + Alert( + AlertSeverity.HIGH, + f"*{asset.name} fundamental oracle delta* {delta:+.4%} (threshold {threshold:.2%})\n" + f"Previous: {prev_rate}\nCurrent: {current_rate}\nOracle: `{oracle.address}`", + PROTOCOL, + channel=asset.channel, + ) + ) + return alerts + + +# --------------------------------------------------------------------------- +# Orchestration +# --------------------------------------------------------------------------- + + +def _build_rate_oracle_abi(function: str) -> list[dict]: + return [ + { + "inputs": [], + "name": function, + "outputs": [{"internalType": "uint256", "name": "", "type": "uint256"}], + "stateMutability": "view", + "type": "function", + } + ] + + +def _monitor_chainlink_assets(client: Web3Client) -> None: + """Check every registry asset that has a Chainlink feed.""" + assets = [a for a in PEGGED_ASSETS if a.chainlink_feed is not None] + if not assets: + return + + needed_targets: set[PegTarget] = set() + for asset in assets: + assert asset.chainlink_feed is not None + needed_targets.add(asset.peg) + needed_targets.add(asset.chainlink_feed.quote) + peg_prices = resolve_peg_prices(needed_targets) + + market_prices = fetch_prices([a.defillama_key for a in assets]) + readings = read_feeds(client, [a.chainlink_feed.address for a in assets]) # type: ignore[union-attr] + now = int(client.eth.get_block("latest")["timestamp"]) + + for asset in assets: + feed = asset.chainlink_feed + assert feed is not None + reading = readings[feed.address] + + prev_round_raw = get_last_value_for_key_from_file(CACHE_FILE, _round_cache_key(feed.address)) + try: + prev_round_id: int | None = int(str(prev_round_raw)) if str(prev_round_raw) != "0" else None + except ValueError: + prev_round_id = None + + obs = OracleObservation( + asset=asset, + reading=reading, + peg_price_usd=peg_prices[asset.peg], + quote_price_usd=peg_prices[feed.quote], + now=now, + market_price_usd=market_prices.get(asset.defillama_key), + prev_round_id=prev_round_id, + ) + + alerts = evaluate_chainlink_asset(obs) + logger.info( + "%s oracle: $%.6f (peg $%.6f, market %s) — %d alert(s)", + asset.name, + obs.oracle_price_usd, + obs.peg_price_usd, + f"${obs.market_price_usd:.6f}" if obs.market_price_usd is not None else "n/a", + len(alerts), + ) + for alert in alerts: + send_alert(alert) + + write_last_value_to_file(CACHE_FILE, _round_cache_key(feed.address), reading.round_data.round_id) + + +def _monitor_rate_oracles(client: Web3Client) -> None: + """Check every registry asset that has a fundamental / rate oracle.""" + assets = [a for a in PEGGED_ASSETS if a.rate_oracle is not None] + for asset in assets: + oracle = asset.rate_oracle + assert oracle is not None + contract = client.get_contract(oracle.address, _build_rate_oracle_abi(oracle.function)) + current_rate = int(contract.functions[oracle.function]().call()) + + prev_raw = get_last_value_for_key_from_file(CACHE_FILE, _rate_cache_key(oracle.address)) + try: + prev_rate: int | None = int(str(prev_raw)) if str(prev_raw) != "0" else None + except ValueError: + prev_rate = None + + alerts = check_rate_oracle(asset, current_rate, prev_rate) + logger.info("%s rate oracle %s: %d (%d alert(s))", asset.name, oracle.address, current_rate, len(alerts)) + for alert in alerts: + send_alert(alert) + + write_last_value_to_file(CACHE_FILE, _rate_cache_key(oracle.address), current_rate) + + +def main() -> None: + """Run all L2 oracle-health checks driven by the pegged-asset registry.""" + client = ChainManager.get_client(Chain.MAINNET) + _monitor_chainlink_assets(client) + _monitor_rate_oracles(client) + logger.info("L2 oracle health check complete (%d Tenderly-covered oracle(s) skipped)", len(TENDERLY_COVERED)) + + +if __name__ == "__main__": + from utils.runner import run_with_alert + + run_with_alert(main, PROTOCOL) diff --git a/tests/test_stables_oracles.py b/tests/test_stables_oracles.py new file mode 100644 index 00000000..41e9654f --- /dev/null +++ b/tests/test_stables_oracles.py @@ -0,0 +1,180 @@ +import unittest +from decimal import Decimal + +from protocols.stables.oracles import ( + OracleObservation, + check_market_divergence, + check_peg_deviation, + check_rate_oracle, + check_round_health, + check_staleness, + evaluate_chainlink_asset, +) +from utils.alert import AlertSeverity +from utils.chainlink import FeedReading, RoundData +from utils.pegged_assets import PeggedAsset, PegTarget, RateOracle, get_asset + +NOW = 2_000_000 +HEARTBEAT = 86_400 # matches registry _STABLE_HEARTBEAT + + +def _reading( + address: str, + answer: int, + *, + decimals: int = 8, + round_id: int = 100, + updated_at: int = NOW - 100, + answered_in_round: int = 100, +) -> FeedReading: + rd = RoundData( + round_id=round_id, + answer=answer, + started_at=updated_at, + updated_at=updated_at, + answered_in_round=answered_in_round, + ) + return FeedReading(address=address, round_data=rd, decimals=decimals) + + +def _cbbtc_obs(**overrides) -> OracleObservation: + """A healthy cbBTC (USD-quoted feed, BTC peg) observation; override per test.""" + asset = get_asset("cbBTC") + defaults = dict( + asset=asset, + reading=_reading(asset.chainlink_feed.address, 60_100 * 10**8), # $60,100 + peg_price_usd=Decimal("60000"), + quote_price_usd=Decimal("1"), # USD-quoted feed + now=NOW, + market_price_usd=Decimal("60100"), + prev_round_id=99, + ) + defaults.update(overrides) + return OracleObservation(**defaults) + + +class TestHealthyAsset(unittest.TestCase): + def test_no_alerts_when_healthy(self): + self.assertEqual(evaluate_chainlink_asset(_cbbtc_obs()), []) + + +class TestStaleness(unittest.TestCase): + def test_fresh_feed_ok(self): + self.assertIsNone(check_staleness(_cbbtc_obs(), buffer=600)) + + def test_forced_stale_fires(self): + stale = _cbbtc_obs( + reading=_reading( + get_asset("cbBTC").chainlink_feed.address, 60_100 * 10**8, updated_at=NOW - (HEARTBEAT + 1000) + ) + ) + alert = check_staleness(stale, buffer=600) + self.assertIsNotNone(alert) + self.assertEqual(alert.severity, AlertSeverity.HIGH) + self.assertEqual(alert.channel, "pegs") + + def test_zero_updated_at_is_stale(self): + obs = _cbbtc_obs(reading=_reading(get_asset("cbBTC").chainlink_feed.address, 60_100 * 10**8, updated_at=0)) + self.assertIsNotNone(check_staleness(obs)) + + +class TestRoundHealth(unittest.TestCase): + def test_healthy_round(self): + self.assertIsNone(check_round_health(_cbbtc_obs())) + + def test_non_positive_answer_is_critical(self): + obs = _cbbtc_obs(reading=_reading(get_asset("cbBTC").chainlink_feed.address, 0)) + alert = check_round_health(obs) + self.assertIsNotNone(alert) + self.assertEqual(alert.severity, AlertSeverity.CRITICAL) + + def test_lagging_answered_in_round_is_high(self): + addr = get_asset("cbBTC").chainlink_feed.address + obs = _cbbtc_obs(reading=_reading(addr, 60_100 * 10**8, round_id=100, answered_in_round=99)) + alert = check_round_health(obs) + self.assertIsNotNone(alert) + self.assertEqual(alert.severity, AlertSeverity.HIGH) + + def test_roundid_backwards_is_critical(self): + obs = _cbbtc_obs(prev_round_id=200) # current round_id is 100 + alert = check_round_health(obs) + self.assertIsNotNone(alert) + self.assertEqual(alert.severity, AlertSeverity.CRITICAL) + + +class TestPegDeviation(unittest.TestCase): + def test_within_tolerance_ok(self): + self.assertIsNone(check_peg_deviation(_cbbtc_obs())) + + def test_off_peg_fires(self): + # oracle $63,000 vs $60,000 peg = +5% > cbBTC 2% tolerance + obs = _cbbtc_obs(reading=_reading(get_asset("cbBTC").chainlink_feed.address, 63_000 * 10**8)) + alert = check_peg_deviation(obs) + self.assertIsNotNone(alert) + self.assertEqual(alert.severity, AlertSeverity.HIGH) + + +class TestMarketDivergence(unittest.TestCase): + def test_aligned_ok(self): + self.assertIsNone(check_market_divergence(_cbbtc_obs(), threshold=Decimal("0.01"))) + + def test_forced_divergence_fires(self): + # oracle $60,100 vs market $50,000 ~ +20% + obs = _cbbtc_obs(market_price_usd=Decimal("50000")) + alert = check_market_divergence(obs, threshold=Decimal("0.01")) + self.assertIsNotNone(alert) + self.assertEqual(alert.severity, AlertSeverity.HIGH) + + def test_missing_market_price_skips(self): + self.assertIsNone(check_market_divergence(_cbbtc_obs(market_price_usd=None))) + + +class TestQuoteConversion(unittest.TestCase): + def test_btc_quoted_feed_scales_to_usd(self): + # LBTC/BTC feed answer 1.004 BTC, BTC at $60,000 -> oracle $60,240 + lbtc = get_asset("LBTC") + obs = OracleObservation( + asset=lbtc, + reading=_reading(lbtc.chainlink_feed.address, 100_400_000), # 1.004 * 1e8 + peg_price_usd=Decimal("60000"), + quote_price_usd=Decimal("60000"), # feed quotes in BTC + now=NOW, + market_price_usd=Decimal("60240"), + ) + self.assertEqual(obs.oracle_price_usd, Decimal("60240.000")) + self.assertEqual(evaluate_chainlink_asset(obs), []) + + +class TestRateOracle(unittest.TestCase): + def _asset(self, monotonic: bool = True) -> PeggedAsset: + return PeggedAsset( + name="fakeRate", + defillama_key="ethereum:0x0000000000000000000000000000000000000001", + channel="pegs", + peg=PegTarget.USD, + depeg_pct=Decimal("0.02"), + rate_oracle=RateOracle(address="0xRate", monotonic=monotonic), + ) + + def test_no_previous_rate_no_alert(self): + self.assertEqual(check_rate_oracle(self._asset(), current_rate=10**18, prev_rate=None), []) + + def test_monotonic_decrease_is_critical(self): + alerts = check_rate_oracle(self._asset(monotonic=True), current_rate=9 * 10**17, prev_rate=10**18) + self.assertEqual(len(alerts), 1) + self.assertEqual(alerts[0].severity, AlertSeverity.CRITICAL) + + def test_large_increase_is_high(self): + alerts = check_rate_oracle(self._asset(), current_rate=12 * 10**17, prev_rate=10**18, threshold=Decimal("0.05")) + self.assertEqual(len(alerts), 1) + self.assertEqual(alerts[0].severity, AlertSeverity.HIGH) + + def test_small_change_no_alert(self): + self.assertEqual( + check_rate_oracle(self._asset(), current_rate=101 * 10**16, prev_rate=10**18, threshold=Decimal("0.05")), + [], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/pegged_assets.py b/utils/pegged_assets.py index a5b77424..ceda2a7f 100644 --- a/utils/pegged_assets.py +++ b/utils/pegged_assets.py @@ -41,11 +41,17 @@ class PegTarget(Enum): @dataclass(frozen=True) class ChainlinkFeed: - """A Chainlink aggregator backing an asset's price (consumed by L2).""" + """A Chainlink aggregator backing an asset's price (consumed by L2). + + ``quote`` is the unit the feed is denominated in (e.g. a ``LBTC/BTC`` feed + quotes in ``BTC``, a ``cbBTC/USD`` feed in ``USD``); L2 multiplies the raw + answer by the quote's live price to compare oracle and market on a USD basis. + """ address: str heartbeat: int # max expected seconds between updates (Chainlink mainnet default) description: str = "" + quote: PegTarget = PegTarget.USD @dataclass(frozen=True) @@ -233,7 +239,9 @@ def get_asset(name: str) -> PeggedAsset: peg=PegTarget.BTC, depeg_pct=Decimal("0.03"), # LBTC/BTC market-rate feed (8 decimals); price ~1 BTC per LBTC. - chainlink_feed=ChainlinkFeed("0x5c29868C58b6e15e2b962943278969Ab6a7D3212", _STABLE_HEARTBEAT, "LBTC/BTC"), + chainlink_feed=ChainlinkFeed( + "0x5c29868C58b6e15e2b962943278969Ab6a7D3212", _STABLE_HEARTBEAT, "LBTC/BTC", quote=PegTarget.BTC + ), ), ] From 210658867e0e2a4f754d8dcb44abd9e73f120c77 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 29 Jun 2026 20:20:38 +0000 Subject: [PATCH 03/10] fix(pegs): logical protocol for dispatch + non-poisoning round cache (#301) Address review on stables/oracles.py: 1. Emergency dispatch (utils/dispatch.dispatch_emergency_withdrawal) keys off Alert.protocol and skips anything outside DISPATCHABLE_PROTOCOLS. Alerts were hardcoding protocol="pegs" and putting the owner only in channel, so an USDe oracle failure routed to Telegram but never triggered ethena's emergency withdrawal. Rename PeggedAsset.channel -> protocol (the logical owner) and add an optional channel override (empty falls back to protocol routing). Alerts now emit protocol=asset.protocol, channel=asset.channel, so dispatch fires for ethena/cap/infinifi while Telegram routing is unchanged. 2. The round cache was overwritten with the current roundId unconditionally, so a backwards roundId became the new baseline and the regression was caught only once. Add pure next_cached_round(): a health-gated high-water mark that never lowers the cached round or stores a malfunctioning one. Tests: add dispatch-routing coverage (alert.protocol in DISPATCHABLE_PROTOCOLS) and next_cached_round cases. Full suite: 637 passed, 6 skipped. ruff + mypy clean (new files). Live no-send smoke reads all six feeds, 0 alerts under current conditions. Co-Authored-By: Claude Opus 4.8 (1M context) --- protocols/stables/oracles.py | 34 +++++++++++++++++++------ tests/test_stables_oracles.py | 47 +++++++++++++++++++++++++++++++++-- utils/pegged_assets.py | 21 ++++++++-------- 3 files changed, 82 insertions(+), 20 deletions(-) diff --git a/protocols/stables/oracles.py b/protocols/stables/oracles.py index f71ba770..53459350 100644 --- a/protocols/stables/oracles.py +++ b/protocols/stables/oracles.py @@ -25,7 +25,7 @@ from utils.alert import Alert, AlertSeverity, send_alert from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file -from utils.chainlink import FeedReading, read_feeds, round_issues +from utils.chainlink import FeedReading, RoundData, is_round_healthy, read_feeds, round_issues from utils.chains import Chain from utils.config import Config from utils.defillama import fetch_prices @@ -103,7 +103,7 @@ def check_staleness(obs: OracleObservation, buffer: int = STALENESS_BUFFER) -> A f"Age: {age}s — heartbeat {feed.heartbeat}s + buffer {buffer}s\n" f"updatedAt: {updated_at}\n" f"Feed: `{feed.address}`", - PROTOCOL, + obs.asset.protocol, channel=obs.asset.channel, ) return None @@ -134,7 +134,7 @@ def check_round_health(obs: OracleObservation) -> Alert | None: f"*{obs.asset.name} oracle round unhealthy* ({feed.description})\n" + "\n".join(f"- {issue}" for issue in issues) + f"\nFeed: `{feed.address}`", - PROTOCOL, + obs.asset.protocol, channel=obs.asset.channel, ) @@ -152,7 +152,7 @@ def check_peg_deviation(obs: OracleObservation) -> Alert | None: f"Oracle: ${obs.oracle_price_usd:.6f}\n" f"Peg: ${obs.peg_price_usd:.6f}\n" f"Deviation: {dev:+.2%} (tolerance {obs.asset.depeg_pct:.2%})", - PROTOCOL, + obs.asset.protocol, channel=obs.asset.channel, ) @@ -170,7 +170,7 @@ def check_market_divergence(obs: OracleObservation, threshold: Decimal = DIVERGE f"Oracle: ${obs.oracle_price_usd:.6f}\n" f"Market (DeFiLlama): ${obs.market_price_usd:.6f}\n" f"Divergence: {dev:+.2%} (threshold {threshold:.2%})", - PROTOCOL, + obs.asset.protocol, channel=obs.asset.channel, ) @@ -191,6 +191,22 @@ def evaluate_chainlink_asset( return [alert for alert in candidates if alert is not None] +def next_cached_round(prev_round_id: int | None, round_data: RoundData) -> int: + """High-water-mark ``roundId`` to persist so a regression never poisons the cache. + + The cached ``roundId`` is the baseline the next run compares against for + monotonicity. Writing a lower or malfunctioning round would make the + regression the new baseline, so it is flagged only once and a feed stuck at + (or crawling below) the regressed round looks "monotonic" forever. Only a + healthy, non-decreasing round advances the mark. + """ + if not is_round_healthy(round_data): + return prev_round_id or 0 + if prev_round_id is None: + return round_data.round_id + return max(prev_round_id, round_data.round_id) + + def check_rate_oracle( asset: PeggedAsset, current_rate: int, @@ -217,7 +233,7 @@ def check_rate_oracle( f"*{asset.name} fundamental oracle DECREASED* (monotonic/capped)\n" f"Previous: {prev_rate}\nCurrent: {current_rate}\nDelta: {delta:+.4%}\n" f"Oracle: `{oracle.address}`", - PROTOCOL, + asset.protocol, channel=asset.channel, ) ) @@ -227,7 +243,7 @@ def check_rate_oracle( AlertSeverity.HIGH, f"*{asset.name} fundamental oracle delta* {delta:+.4%} (threshold {threshold:.2%})\n" f"Previous: {prev_rate}\nCurrent: {current_rate}\nOracle: `{oracle.address}`", - PROTOCOL, + asset.protocol, channel=asset.channel, ) ) @@ -301,7 +317,9 @@ def _monitor_chainlink_assets(client: Web3Client) -> None: for alert in alerts: send_alert(alert) - write_last_value_to_file(CACHE_FILE, _round_cache_key(feed.address), reading.round_data.round_id) + write_last_value_to_file( + CACHE_FILE, _round_cache_key(feed.address), next_cached_round(prev_round_id, reading.round_data) + ) def _monitor_rate_oracles(client: Web3Client) -> None: diff --git a/tests/test_stables_oracles.py b/tests/test_stables_oracles.py index 41e9654f..8a7283df 100644 --- a/tests/test_stables_oracles.py +++ b/tests/test_stables_oracles.py @@ -9,9 +9,11 @@ check_round_health, check_staleness, evaluate_chainlink_asset, + next_cached_round, ) from utils.alert import AlertSeverity from utils.chainlink import FeedReading, RoundData +from utils.dispatch import DISPATCHABLE_PROTOCOLS from utils.pegged_assets import PeggedAsset, PegTarget, RateOracle, get_asset NOW = 2_000_000 @@ -71,7 +73,9 @@ def test_forced_stale_fires(self): alert = check_staleness(stale, buffer=600) self.assertIsNotNone(alert) self.assertEqual(alert.severity, AlertSeverity.HIGH) - self.assertEqual(alert.channel, "pegs") + # cbBTC has no dispatchable owner: protocol is "pegs", channel override empty. + self.assertEqual(alert.protocol, "pegs") + self.assertEqual(alert.channel, "") def test_zero_updated_at_is_stale(self): obs = _cbbtc_obs(reading=_reading(get_asset("cbBTC").chainlink_feed.address, 60_100 * 10**8, updated_at=0)) @@ -150,7 +154,7 @@ def _asset(self, monotonic: bool = True) -> PeggedAsset: return PeggedAsset( name="fakeRate", defillama_key="ethereum:0x0000000000000000000000000000000000000001", - channel="pegs", + protocol="pegs", peg=PegTarget.USD, depeg_pct=Decimal("0.02"), rate_oracle=RateOracle(address="0xRate", monotonic=monotonic), @@ -176,5 +180,44 @@ def test_small_change_no_alert(self): ) +class TestDispatchRouting(unittest.TestCase): + """Alerts must carry the asset's logical protocol so emergency dispatch can fire.""" + + def test_owned_asset_uses_dispatchable_protocol(self): + usde = get_asset("USDe") # owner "ethena", peg USD, 3% tolerance + obs = OracleObservation( + asset=usde, + reading=_reading(usde.chainlink_feed.address, 90 * 10**6), # $0.90 -> off peg + peg_price_usd=Decimal("1"), + quote_price_usd=Decimal("1"), + now=NOW, + market_price_usd=Decimal("0.90"), + ) + alert = check_peg_deviation(obs) + self.assertIsNotNone(alert) + # protocol (not channel) carries the owner; dispatch keys off alert.protocol. + self.assertEqual(alert.protocol, "ethena") + self.assertIn(alert.protocol, DISPATCHABLE_PROTOCOLS) + + +class TestNextCachedRound(unittest.TestCase): + def _rd(self, round_id: int, answer: int = 60_100 * 10**8, updated_at: int = NOW - 100) -> RoundData: + return RoundData(round_id, answer, updated_at, updated_at, round_id) + + def test_first_run_caches_current(self): + self.assertEqual(next_cached_round(None, self._rd(100)), 100) + + def test_advances_on_increase(self): + self.assertEqual(next_cached_round(100, self._rd(101)), 101) + + def test_keeps_high_water_mark_on_regression(self): + # Backwards round must NOT lower the cached baseline (no poisoning). + self.assertEqual(next_cached_round(100, self._rd(99)), 100) + + def test_broken_round_does_not_poison_even_if_higher(self): + # answer == 0 -> unhealthy; keep last-good rather than caching a broken round. + self.assertEqual(next_cached_round(100, self._rd(200, answer=0)), 100) + + if __name__ == "__main__": unittest.main() diff --git a/utils/pegged_assets.py b/utils/pegged_assets.py index ceda2a7f..6869f676 100644 --- a/utils/pegged_assets.py +++ b/utils/pegged_assets.py @@ -79,11 +79,12 @@ class PeggedAsset: name: str defillama_key: str # "chain:address" - channel: str # Telegram routing channel + protocol: str # logical owner — used as Alert.protocol so emergency dispatch can key off it peg: PegTarget depeg_pct: Decimal # deviation tolerance from the peg (e.g. Decimal("0.02") = 2%) chainlink_feed: ChainlinkFeed | None = None rate_oracle: RateOracle | None = None + channel: str = "" # Telegram channel override; empty falls back to ``protocol`` routing @property def address(self) -> str: @@ -171,7 +172,7 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="USDC", defillama_key="ethereum:0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", - channel="pegs", + protocol="pegs", peg=PegTarget.USD, depeg_pct=Decimal("0.02"), chainlink_feed=ChainlinkFeed("0x8fFfFfd4AfB6115b954Bd326cbe7B4BA576818f6", _STABLE_HEARTBEAT, "USDC/USD"), @@ -179,7 +180,7 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="USDT", defillama_key="ethereum:0xdAC17F958D2ee523a2206206994597C13D831ec7", - channel="pegs", + protocol="pegs", peg=PegTarget.USD, depeg_pct=Decimal("0.02"), chainlink_feed=ChainlinkFeed("0x3E7d1eAB13ad0104d2750B8863b489D65364e32D", _STABLE_HEARTBEAT, "USDT/USD"), @@ -187,7 +188,7 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="USDS", defillama_key="ethereum:0xdC035D45d973E3EC169d2276DDab16f1e407384F", - channel="pegs", + protocol="pegs", peg=PegTarget.USD, depeg_pct=Decimal("0.02"), chainlink_feed=ChainlinkFeed("0xfF30586cD0F29eD462364C7e81375FC0C71219b1", _STABLE_HEARTBEAT, "USDS/USD"), @@ -196,7 +197,7 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="USDe", defillama_key="ethereum:0x4c9EDD5852cd905f086C759E8383e09bff1E68B3", - channel="ethena", + protocol="ethena", peg=PegTarget.USD, depeg_pct=Decimal("0.03"), chainlink_feed=ChainlinkFeed("0xa569d910839Ae8865Da8F8e70FfFb0cBA869F961", _STABLE_HEARTBEAT, "USDe/USD"), @@ -204,14 +205,14 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="cUSD", defillama_key="ethereum:0xcccc62962d17b8914c62d74ffb843d73b2a3cccc", - channel="cap", + protocol="cap", peg=PegTarget.USD, depeg_pct=Decimal("0.05"), # cap cUSD price is more volatile ), PeggedAsset( name="iUSD", defillama_key="ethereum:0x48f9e38f3070AD8945DFEae3FA70987722E3D89c", - channel="infinifi", + protocol="infinifi", peg=PegTarget.USD, depeg_pct=Decimal("0.03"), ), @@ -219,7 +220,7 @@ def get_asset(name: str) -> PeggedAsset: # TODO: siUSD token address not yet verified on-chain — placeholder. name="siUSD", defillama_key=f"ethereum:{PLACEHOLDER_ADDRESS}", - channel="pegs", + protocol="pegs", peg=PegTarget.USD, depeg_pct=Decimal("0.05"), ), @@ -227,7 +228,7 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="cbBTC", defillama_key="ethereum:0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf", - channel="pegs", + protocol="pegs", peg=PegTarget.BTC, depeg_pct=Decimal("0.02"), chainlink_feed=ChainlinkFeed("0x2665701293fCbEB223D11A08D826563EDcCE423A", _STABLE_HEARTBEAT, "cbBTC/USD"), @@ -235,7 +236,7 @@ def get_asset(name: str) -> PeggedAsset: PeggedAsset( name="LBTC", defillama_key="ethereum:0x8236a87084f8B84306f72007F36F2618A5634494", - channel="pegs", + protocol="pegs", peg=PegTarget.BTC, depeg_pct=Decimal("0.03"), # LBTC/BTC market-rate feed (8 decimals); price ~1 BTC per LBTC. From 7a6e01e06328464e0e9d7f0f306f0dc238096e2b Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 29 Jun 2026 21:11:12 +0000 Subject: [PATCH 04/10] =?UTF-8?q?feat(pegs):=20L3=20oracle-events=20consum?= =?UTF-8?q?er=20=E2=80=94=20stables/oracle=5Fevents.py=20(#302)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Layer 3 of peg monitoring: an Envio GraphQL consumer that reads Chainlink AnswerUpdated rows and turns per-feed anomalies into alerts. Stacks on L2 (#301) for the shared registry routing. Mirrors protocols/timelock/timelock_alerts.py. Detects per feed (pure, unit-tested functions): - large round-over-round jumps (|Δanswer|/prev >= JUMP_THRESHOLD, default 10%), - missed-heartbeat gaps (updatedAt gap > feed heartbeat + buffer), - sequence anomalies (roundId not strictly increasing -> CRITICAL). Routing uses PeggedAsset.protocol + channel so alerts reach the owning protocol and its emergency dispatch (consistent with the L2 fix). De-dupe via a per-aggregator blockTimestamp cursor, advanced only when every send lands (same trade-off as timelock_alerts: re-alert on retry is acceptable, dropping is not). Address sourcing (per indexer issue chain-events/yearn-indexing-test#31): AnswerUpdated is emitted by the underlying aggregator (rotates on phase upgrades), which Envio can't scope by feed. The consumer resolves each feed proxy's current aggregator() on-chain (batched) and maps aggregator -> asset, so it always tracks the live aggregator. GraphQL field names are centralised in _F_* constants to align with #31's final schema. Wired into the hourly profile (renders; 25 tasks). ruff + mypy clean (new files); 651 passed, 6 skipped. Live smoke resolved all 6 aggregators on-chain and handled the not-yet-indexed entity gracefully via send_error_message. Co-Authored-By: Claude Opus 4.8 (1M context) --- automation/jobs.yaml | 1 + protocols/stables/oracle_events.py | 390 ++++++++++++++++++++++++++++ tests/test_stables_oracle_events.py | 141 ++++++++++ 3 files changed, 532 insertions(+) create mode 100644 protocols/stables/oracle_events.py create mode 100644 tests/test_stables_oracle_events.py diff --git a/automation/jobs.yaml b/automation/jobs.yaml index 892b6b1b..ecce5439 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -38,6 +38,7 @@ profiles: - { name: "usdai-large-mints", script: protocols/usdai/large_mints.py } - { name: "stables-dune-large-transfers", script: protocols/stables/dune_large_transfers.py } - { name: "stables-oracles", script: protocols/stables/oracles.py } + - { name: "stables-oracle-events", script: protocols/stables/oracle_events.py } - { name: "yearn-alert-large-flows", script: protocols/yearn/alert_large_flows.py } # Cache: tks-trigger-cache.json under $CACHE_DIR (check_stuck_triggers.DEFAULT_CACHE_FILE). - { name: "yearn-check-stuck-triggers", script: protocols/yearn/check_stuck_triggers.py, enabled: false } diff --git a/protocols/stables/oracle_events.py b/protocols/stables/oracle_events.py new file mode 100644 index 00000000..d192d1d5 --- /dev/null +++ b/protocols/stables/oracle_events.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +"""Layer 3 peg monitoring — Chainlink ``AnswerUpdated`` event consumer (hourly). + +Reads the Chainlink ``AnswerUpdated(int256 current, uint256 roundId, uint256 updatedAt)`` +rows captured by the Envio indexer (``chain-events/yearn-indexing-test`` issue #31) +and turns per-feed anomalies into alerts. Where L2 (``oracles.py``) polls the +*current* round each hour, this consumes the *full event stream* so no round is +missed between polls. Mirrors the Envio→Telegram pattern of +``protocols/timelock/timelock_alerts.py``. + +Detects, per feed: + +* **large round-over-round jumps** — ``|Δanswer| / prev ≥ JUMP_THRESHOLD``; +* **missed-heartbeat gaps** — ``updatedAt`` gap between consecutive rounds + exceeds the feed heartbeat + buffer; +* **sequence anomalies** — a ``roundId`` that does not strictly increase. + +Routing uses the shared :data:`PEGGED_ASSETS` registry (``protocol`` + ``channel``) +so alerts reach the owning protocol and its emergency dispatch. De-duped across +runs via a per-aggregator ``blockTimestamp`` cursor in the cache. + +Address sourcing (per indexer #31): ``AnswerUpdated`` is emitted by the *underlying +aggregator*, whose address we resolve at runtime from each feed proxy's +``aggregator()`` — this also tracks phase rotations (a rotation shows up as a +staleness gap in L2). +""" + +import argparse +import json +import os +import time +import urllib.error +import urllib.request +from dataclasses import dataclass + +from dotenv import load_dotenv +from eth_utils import to_checksum_address + +from utils.alert import Alert, AlertSeverity, send_alert +from utils.cache import cache_filename, get_last_value_for_key_from_file, write_last_value_to_file +from utils.chains import Chain +from utils.config import Config +from utils.logger import get_logger +from utils.pegged_assets import PEGGED_ASSETS, ChainlinkFeed, PeggedAsset +from utils.telegram import send_error_message +from utils.web3_wrapper import ChainManager, Web3Client + +load_dotenv() + +PROTOCOL = "pegs" +_logger = get_logger("stables-oracle-events") + +ENVIO_GRAPHQL_URL = os.getenv("ENVIO_GRAPHQL_URL") + +# Tunables (env-overridable). +JUMP_THRESHOLD = Config.get_env_float("PEG_EVENT_JUMP_THRESHOLD", 0.10) # 10% round-over-round +HEARTBEAT_BUFFER = Config.get_env_int("PEG_EVENT_HEARTBEAT_BUFFER", 600) # grace on heartbeat for gap detection +FALLBACK_LOOKBACK = Config.get_env_int("PEG_EVENT_FALLBACK_LOOKBACK", 86_400) # 24h when no cursor yet +QUERY_LIMIT = Config.get_env_int("PEG_EVENT_QUERY_LIMIT", 1000) + +# Extra history fetched before a feed's cursor so the first new round has a prior +# round to diff against (must exceed the largest feed heartbeat). +CONTEXT_WINDOW = Config.get_env_int("PEG_EVENT_CONTEXT_WINDOW", 2 * 86_400) # 2 days + +# Minimal EACAggregatorProxy ABI — resolve the underlying aggregator that emits AnswerUpdated. +AGGREGATOR_PROXY_ABI = [ + { + "inputs": [], + "name": "aggregator", + "outputs": [{"internalType": "address", "name": "", "type": "address"}], + "stateMutability": "view", + "type": "function", + } +] + + +def _cursor_key(aggregator: str) -> str: + return f"peg_oracle_event_ts_{aggregator.lower()}" + + +# --------------------------------------------------------------------------- +# Data model + pure anomaly detection (unit tested without a chain or indexer) +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class OracleRound: + """One decoded ``AnswerUpdated`` row.""" + + aggregator: str # underlying aggregator address (lowercase) + round_id: int + answer: int + updated_at: int # on-chain updatedAt (unix seconds) + block_timestamp: int # indexer block time (unix seconds) — used for the cursor + tx_hash: str + chain_id: int + + +# GraphQL field names for the AnswerUpdated entity (indexer #31). Centralised so a +# final-schema rename only touches one place. +_F_AGGREGATOR = "aggregatorAddress" +_F_ROUND_ID = "roundId" +_F_ANSWER = "current" +_F_UPDATED_AT = "updatedAt" +_F_BLOCK_TS = "blockTimestamp" +_F_TX = "transactionHash" +_F_CHAIN = "chainId" + + +def parse_round(row: dict) -> OracleRound: + """Map a GraphQL ``AnswerUpdated`` row to an :class:`OracleRound`.""" + return OracleRound( + aggregator=str(row[_F_AGGREGATOR]).lower(), + round_id=int(row[_F_ROUND_ID]), + answer=int(row[_F_ANSWER]), + updated_at=int(row[_F_UPDATED_AT]), + block_timestamp=int(row[_F_BLOCK_TS]), + tx_hash=str(row.get(_F_TX, "")), + chain_id=int(row.get(_F_CHAIN, 1)), + ) + + +def detect_anomalies( + asset: PeggedAsset, + feed: ChainlinkFeed, + rounds: list[OracleRound], + *, + since_ts: int, + jump_threshold: float = JUMP_THRESHOLD, + heartbeat_buffer: int = HEARTBEAT_BUFFER, +) -> list[Alert]: + """Return alerts for anomalies among ``rounds``, only for rounds newer than ``since_ts``. + + ``rounds`` may include up to one round at/just before ``since_ts`` for context + (jump / gap diffing); alerts only fire for the rounds whose ``block_timestamp`` + is strictly greater than ``since_ts`` so reruns never re-alert the same round. + """ + ordered = sorted(rounds, key=lambda r: (r.block_timestamp, r.round_id)) + alerts: list[Alert] = [] + + for prev, cur in zip(ordered, ordered[1:]): + if cur.block_timestamp <= since_ts: + continue # already processed in a prior run + + # Sequence anomaly: roundId must strictly increase. + if cur.round_id <= prev.round_id: + alerts.append( + _alert( + asset, + feed, + AlertSeverity.CRITICAL, + "sequence anomaly — roundId did not increase", + f"Previous roundId: {prev.round_id}\nCurrent roundId: {cur.round_id}", + cur, + ) + ) + + # Large round-over-round jump (unit-independent percentage). + if prev.answer > 0: + jump = abs(cur.answer - prev.answer) / prev.answer + if jump >= jump_threshold: + direction = "up" if cur.answer > prev.answer else "down" + alerts.append( + _alert( + asset, + feed, + AlertSeverity.HIGH, + f"large round-over-round jump {direction} {jump:.2%} (threshold {jump_threshold:.0%})", + f"Previous answer: {prev.answer}\nCurrent answer: {cur.answer}", + cur, + ) + ) + + # Missed-heartbeat gap between consecutive updates. + gap = cur.updated_at - prev.updated_at + if gap > feed.heartbeat + heartbeat_buffer: + alerts.append( + _alert( + asset, + feed, + AlertSeverity.HIGH, + f"missed-heartbeat gap {gap}s (heartbeat {feed.heartbeat}s + buffer {heartbeat_buffer}s)", + f"Previous updatedAt: {prev.updated_at}\nCurrent updatedAt: {cur.updated_at}", + cur, + ) + ) + + return alerts + + +def _alert( + asset: PeggedAsset, feed: ChainlinkFeed, severity: AlertSeverity, summary: str, detail: str, cur: OracleRound +) -> Alert: + """Build a routed alert for a feed anomaly.""" + tx_line = f"\nTx: `{cur.tx_hash}`" if cur.tx_hash else "" + return Alert( + severity, + f"*{asset.name} oracle event — {summary}* ({feed.description})\n" + f"{detail}\n" + f"Aggregator: `{cur.aggregator}`" + f"{tx_line}", + asset.protocol, + channel=asset.channel, + ) + + +def next_cursor(prev_cursor: int, rounds: list[OracleRound]) -> int: + """Highest ``block_timestamp`` seen, but never below ``prev_cursor``.""" + return max([prev_cursor, *(r.block_timestamp for r in rounds)]) + + +# --------------------------------------------------------------------------- +# Envio GraphQL (mirrors timelock_alerts.py) +# --------------------------------------------------------------------------- + + +def http_json(url: str, method: str = "GET", body: dict | None = None, headers: dict | None = None) -> dict | None: + """Make an HTTP request and return the JSON response, retrying transient errors.""" + _logger.info("http_json %s %s", method, url) + data = None + req_headers: dict[str, str] = {"Accept": "application/json"} + if headers: + req_headers.update(headers) + if body is not None: + data = json.dumps(body).encode("utf-8") + req_headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=req_headers, method=method) + max_retries = 3 + for attempt in range(1, max_retries + 1): + try: + with urllib.request.urlopen(req, timeout=30) as resp: + parsed = json.loads(resp.read().decode("utf-8")) + return parsed if isinstance(parsed, dict) else None + except (urllib.error.URLError, ConnectionError, OSError) as e: + _logger.warning("http_json attempt %s/%s failed: %s", attempt, max_retries, e) + if attempt < max_retries: + time.sleep(2 * attempt) + return None + + +def gql_request(query: str, variables: dict) -> dict | None: + """Execute a GraphQL query against the Envio indexer.""" + if not ENVIO_GRAPHQL_URL: + raise RuntimeError( + "ENVIO_GRAPHQL_URL is not set. Set it to the Envio GraphQL endpoint, " + "e.g. export ENVIO_GRAPHQL_URL='https://.../graphql'." + ) + return http_json(ENVIO_GRAPHQL_URL, method="POST", body={"query": query, "variables": variables}) + + +def load_answer_updated(aggregators: list[str], since_ts: int, limit: int) -> dict | None: + """Fetch ``AnswerUpdated`` rows for the given aggregator addresses since ``since_ts``.""" + addresses = sorted({addr for agg in aggregators for addr in (agg, to_checksum_address(agg))}) + query = """ + query GetAnswerUpdated($limit: Int!, $sinceTs: Int!, $addresses: [String!]!) { + AnswerUpdated( + where: { + aggregatorAddress: { _in: $addresses } + blockTimestamp: { _gt: $sinceTs } + } + order_by: { blockTimestamp: asc, logIndex: asc } + limit: $limit + ) { + id + aggregatorAddress + roundId + current + updatedAt + blockNumber + blockTimestamp + transactionHash + chainId + } + } + """ + return gql_request(query, {"limit": limit, "sinceTs": since_ts, "addresses": addresses}) + + +# --------------------------------------------------------------------------- +# Orchestration +# --------------------------------------------------------------------------- + + +def resolve_aggregators(client: Web3Client, assets: list[PeggedAsset]) -> dict[str, tuple[PeggedAsset, ChainlinkFeed]]: + """Resolve each feed proxy's current underlying aggregator (lowercase) → (asset, feed).""" + feeds = [(a, a.chainlink_feed) for a in assets if a.chainlink_feed is not None] + contracts = [client.get_contract(feed.address, AGGREGATOR_PROXY_ABI) for _, feed in feeds] + + with client.batch_requests() as batch: + for contract in contracts: + batch.add(contract.functions.aggregator()) + responses = client.execute_batch(batch) + + mapping: dict[str, tuple[PeggedAsset, ChainlinkFeed]] = {} + for (asset, feed), aggregator in zip(feeds, responses): + agg = str(aggregator).lower() + mapping[agg] = (asset, feed) + _logger.info("%s feed %s -> aggregator %s", asset.name, feed.address, agg) + return mapping + + +def process_rounds( + aggregator_map: dict[str, tuple[PeggedAsset, ChainlinkFeed]], + rows: list[dict], + use_cache: bool, +) -> None: + """Group rows per aggregator, detect anomalies, alert, and advance per-feed cursors.""" + by_aggregator: dict[str, list[OracleRound]] = {} + for row in rows: + rnd = parse_round(row) + if rnd.aggregator in aggregator_map: + by_aggregator.setdefault(rnd.aggregator, []).append(rnd) + + for aggregator, (asset, feed) in aggregator_map.items(): + rounds = by_aggregator.get(aggregator, []) + if not rounds: + continue + + cursor = _read_cursor(aggregator) if use_cache else 0 + alerts = detect_anomalies(asset, feed, rounds, since_ts=cursor) + _logger.info( + "%s (%s): %d round(s), %d alert(s) since cursor %s", + asset.name, + feed.description, + len(rounds), + len(alerts), + cursor, + ) + + all_sent = True + for alert in alerts: + try: + send_alert(alert) + except Exception: + _logger.exception("Failed to send alert for %s", asset.name) + all_sent = False + + # Advance the cursor only when every send landed, so a failed send is retried + # next run rather than silently skipped (same trade-off as timelock_alerts). + if use_cache and all_sent: + new_cursor = next_cursor(cursor, rounds) + if new_cursor > cursor: + write_last_value_to_file(cache_filename, _cursor_key(aggregator), str(new_cursor)) + + +def _read_cursor(aggregator: str) -> int: + cached = get_last_value_for_key_from_file(cache_filename, _cursor_key(aggregator)) + if cached and str(cached) != "0": + return int(str(cached)) + return int(time.time()) - FALLBACK_LOOKBACK + + +def main() -> None: + parser = argparse.ArgumentParser(description="Alert on Chainlink AnswerUpdated anomalies.") + parser.add_argument("--limit", type=int, default=QUERY_LIMIT) + parser.add_argument("--no-cache", action="store_true", help="Disable cursor caching (re-scan the window).") + args = parser.parse_args() + use_cache = not args.no_cache + + assets = [a for a in PEGGED_ASSETS if a.chainlink_feed is not None] + if not assets: + _logger.info("No Chainlink-backed assets in registry; nothing to do") + return + + client = ChainManager.get_client(Chain.MAINNET) + aggregator_map = resolve_aggregators(client, assets) + + # Single query covering all feeds: earliest cursor minus the context window so + # each feed gets a prior round to diff against; per-feed gating happens later. + cursors = [_read_cursor(agg) if use_cache else 0 for agg in aggregator_map] + since_ts = max(0, min(cursors) - CONTEXT_WINDOW) + _logger.info("Querying AnswerUpdated for %d aggregators since %s", len(aggregator_map), since_ts) + + response = load_answer_updated(list(aggregator_map), since_ts, args.limit) + if response is None: + send_error_message("Peg oracle events: Envio API unreachable after 3 retries", PROTOCOL) + return + if "errors" in response: + send_error_message(f"Peg oracle events: GraphQL errors: {response['errors']}", PROTOCOL) + return + + rows = response.get("data", {}).get("AnswerUpdated", []) + _logger.info("Fetched %d AnswerUpdated rows", len(rows)) + process_rounds(aggregator_map, rows, use_cache) + + +if __name__ == "__main__": + from utils.runner import run_with_alert + + run_with_alert(main, PROTOCOL) diff --git a/tests/test_stables_oracle_events.py b/tests/test_stables_oracle_events.py new file mode 100644 index 00000000..f6a816b5 --- /dev/null +++ b/tests/test_stables_oracle_events.py @@ -0,0 +1,141 @@ +import unittest + +from protocols.stables.oracle_events import ( + OracleRound, + detect_anomalies, + next_cursor, + parse_round, +) +from utils.alert import AlertSeverity +from utils.dispatch import DISPATCHABLE_PROTOCOLS +from utils.pegged_assets import get_asset + +USDE = get_asset("USDe") # protocol "ethena" (dispatchable), USD/USD feed, 24h heartbeat +FEED = USDE.chainlink_feed +AGG = "0xaggregator" +HB = FEED.heartbeat # 86_400 + + +def _round(round_id: int, answer: int, updated_at: int, *, block_ts: int | None = None) -> OracleRound: + return OracleRound( + aggregator=AGG, + round_id=round_id, + answer=answer, + updated_at=updated_at, + block_timestamp=block_ts if block_ts is not None else updated_at, + tx_hash="0xtx", + chain_id=1, + ) + + +def _detect(rounds, since_ts=0, **kw): + return detect_anomalies(USDE, FEED, rounds, since_ts=since_ts, **kw) + + +class TestNoAnomaly(unittest.TestCase): + def test_healthy_stream_is_quiet(self): + rounds = [ + _round(100, 100_000_000, 1_000), + _round(101, 100_010_000, 1_000 + HB - 10), # small move, within heartbeat + ] + self.assertEqual(_detect(rounds), []) + + def test_single_round_has_no_pair(self): + self.assertEqual(_detect([_round(100, 100_000_000, 1_000)]), []) + + +class TestJump(unittest.TestCase): + def test_large_jump_fires_high(self): + rounds = [_round(100, 100_000_000, 1_000), _round(101, 120_000_000, 1_500)] # +20% + alerts = _detect(rounds, jump_threshold=0.10) + self.assertEqual(len(alerts), 1) + self.assertEqual(alerts[0].severity, AlertSeverity.HIGH) + self.assertIn("jump", alerts[0].message) + + def test_small_move_below_threshold_quiet(self): + rounds = [_round(100, 100_000_000, 1_000), _round(101, 105_000_000, 1_500)] # +5% + self.assertEqual(_detect(rounds, jump_threshold=0.10), []) + + +class TestHeartbeatGap(unittest.TestCase): + def test_gap_beyond_heartbeat_fires(self): + rounds = [_round(100, 100_000_000, 1_000), _round(101, 100_000_001, 1_000 + HB + 5_000)] + alerts = _detect(rounds, heartbeat_buffer=600) + self.assertEqual(len(alerts), 1) + self.assertEqual(alerts[0].severity, AlertSeverity.HIGH) + self.assertIn("missed-heartbeat", alerts[0].message) + + def test_within_heartbeat_quiet(self): + rounds = [_round(100, 100_000_000, 1_000), _round(101, 100_000_001, 1_000 + HB)] + self.assertEqual(_detect(rounds, heartbeat_buffer=600), []) + + +class TestSequence(unittest.TestCase): + def test_non_increasing_round_id_is_critical(self): + # newer block carries a roundId that did not advance + rounds = [_round(101, 100_000_000, 1_000), _round(101, 100_000_000, 1_500)] + alerts = _detect(rounds) + self.assertTrue(any(a.severity == AlertSeverity.CRITICAL for a in alerts)) + self.assertTrue(any("sequence anomaly" in a.message for a in alerts)) + + +class TestDedup(unittest.TestCase): + def test_rounds_at_or_before_cursor_not_realerted(self): + # The jump happens between round 100->101; both at/below the cursor. + rounds = [_round(100, 100_000_000, 1_000), _round(101, 130_000_000, 1_500)] + self.assertEqual(_detect(rounds, since_ts=1_500), []) # cur block_ts == cursor -> skipped + + def test_only_new_rounds_alert_with_prior_context(self): + # round 101 is context (<= cursor); the anomaly on the new round 102 fires once. + rounds = [ + _round(101, 100_000_000, 1_500), + _round(102, 130_000_000, 2_000), # +30% on the new round + ] + alerts = _detect(rounds, since_ts=1_500) + self.assertEqual(len(alerts), 1) + self.assertIn("jump", alerts[0].message) + + +class TestRouting(unittest.TestCase): + def test_alert_uses_dispatchable_protocol(self): + rounds = [_round(100, 100_000_000, 1_000), _round(101, 130_000_000, 1_500)] + alert = _detect(rounds)[0] + self.assertEqual(alert.protocol, "ethena") + self.assertIn(alert.protocol, DISPATCHABLE_PROTOCOLS) + self.assertEqual(alert.channel, "") + + +class TestCursor(unittest.TestCase): + def test_advances_to_max_block_timestamp(self): + rounds = [_round(100, 1, 1_000), _round(101, 1, 2_500)] + self.assertEqual(next_cursor(1_200, rounds), 2_500) + + def test_never_regresses_below_prev(self): + rounds = [_round(100, 1, 1_000)] + self.assertEqual(next_cursor(5_000, rounds), 5_000) + + def test_empty_rounds_keeps_prev(self): + self.assertEqual(next_cursor(5_000, []), 5_000) + + +class TestParseRound(unittest.TestCase): + def test_maps_graphql_row(self): + row = { + "aggregatorAddress": "0xABCDEF", + "roundId": "42", + "current": "99980000", + "updatedAt": "1700000000", + "blockTimestamp": "1700000005", + "transactionHash": "0xdead", + "chainId": 1, + } + rnd = parse_round(row) + self.assertEqual(rnd.aggregator, "0xabcdef") # lowercased + self.assertEqual(rnd.round_id, 42) + self.assertEqual(rnd.answer, 99_980_000) + self.assertEqual(rnd.updated_at, 1_700_000_000) + self.assertEqual(rnd.block_timestamp, 1_700_000_005) + + +if __name__ == "__main__": + unittest.main() From d392e35bdc8a63a566e167d856ce24e5e27dd710 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Mon, 29 Jun 2026 21:32:57 +0000 Subject: [PATCH 05/10] fix(pegs): order oracle events by (blockNumber, logIndex), not roundId (#302) detect_anomalies sorted rounds by (block_timestamp, round_id). round_id is the field whose monotonicity we validate, so using it as the sort tiebreaker reorders same-block events into apparent monotonicity and hides a backwards round: 102 followed by 101 at the same block_timestamp sorted to 101->102 and produced 0 alerts. Carry blockNumber + logIndex through parse_round (and select logIndex in the GraphQL query), and sort by OracleRound.event_order = (block_number, log_index), the canonical on-chain emission order. block_timestamp remains the dedup cursor. Add a regression test: a backwards round sharing a block_timestamp, distinguished only by logIndex, now produces a CRITICAL sequence-anomaly alert. Full suite: 652 passed, 6 skipped. ruff + mypy clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- protocols/stables/oracle_events.py | 22 ++++++++++++++++++++- tests/test_stables_oracle_events.py | 30 +++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/protocols/stables/oracle_events.py b/protocols/stables/oracle_events.py index d192d1d5..6a40b133 100644 --- a/protocols/stables/oracle_events.py +++ b/protocols/stables/oracle_events.py @@ -92,9 +92,21 @@ class OracleRound: answer: int updated_at: int # on-chain updatedAt (unix seconds) block_timestamp: int # indexer block time (unix seconds) — used for the cursor + block_number: int # canonical event order (with log_index); see event_order + log_index: int tx_hash: str chain_id: int + @property + def event_order(self) -> tuple[int, int]: + """Canonical on-chain ordering key — the actual emission order of the event. + + Never order the stream by ``round_id``: that is the very field whose + monotonicity we validate, so using it as a sort key would hide a backwards + round when two events land in the same block. + """ + return (self.block_number, self.log_index) + # GraphQL field names for the AnswerUpdated entity (indexer #31). Centralised so a # final-schema rename only touches one place. @@ -103,6 +115,8 @@ class OracleRound: _F_ANSWER = "current" _F_UPDATED_AT = "updatedAt" _F_BLOCK_TS = "blockTimestamp" +_F_BLOCK_NUM = "blockNumber" +_F_LOG_INDEX = "logIndex" _F_TX = "transactionHash" _F_CHAIN = "chainId" @@ -115,6 +129,8 @@ def parse_round(row: dict) -> OracleRound: answer=int(row[_F_ANSWER]), updated_at=int(row[_F_UPDATED_AT]), block_timestamp=int(row[_F_BLOCK_TS]), + block_number=int(row[_F_BLOCK_NUM]), + log_index=int(row[_F_LOG_INDEX]), tx_hash=str(row.get(_F_TX, "")), chain_id=int(row.get(_F_CHAIN, 1)), ) @@ -135,7 +151,10 @@ def detect_anomalies( (jump / gap diffing); alerts only fire for the rounds whose ``block_timestamp`` is strictly greater than ``since_ts`` so reruns never re-alert the same round. """ - ordered = sorted(rounds, key=lambda r: (r.block_timestamp, r.round_id)) + # Sort by the true emission order (blockNumber, logIndex), NOT round_id — see + # OracleRound.event_order. Using round_id would mask a backwards round whenever + # two updates share a block_timestamp. + ordered = sorted(rounds, key=lambda r: r.event_order) alerts: list[Alert] = [] for prev, cur in zip(ordered, ordered[1:]): @@ -268,6 +287,7 @@ def load_answer_updated(aggregators: list[str], since_ts: int, limit: int) -> di updatedAt blockNumber blockTimestamp + logIndex transactionHash chainId } diff --git a/tests/test_stables_oracle_events.py b/tests/test_stables_oracle_events.py index f6a816b5..e5f383a7 100644 --- a/tests/test_stables_oracle_events.py +++ b/tests/test_stables_oracle_events.py @@ -16,13 +16,24 @@ HB = FEED.heartbeat # 86_400 -def _round(round_id: int, answer: int, updated_at: int, *, block_ts: int | None = None) -> OracleRound: +def _round( + round_id: int, + answer: int, + updated_at: int, + *, + block_ts: int | None = None, + block_number: int | None = None, + log_index: int = 0, +) -> OracleRound: + bt = block_ts if block_ts is not None else updated_at return OracleRound( aggregator=AGG, round_id=round_id, answer=answer, updated_at=updated_at, - block_timestamp=block_ts if block_ts is not None else updated_at, + block_timestamp=bt, + block_number=block_number if block_number is not None else bt, + log_index=log_index, tx_hash="0xtx", chain_id=1, ) @@ -78,6 +89,18 @@ def test_non_increasing_round_id_is_critical(self): self.assertTrue(any(a.severity == AlertSeverity.CRITICAL for a in alerts)) self.assertTrue(any("sequence anomaly" in a.message for a in alerts)) + def test_backwards_round_in_same_block_is_detected(self): + # 102 then 101 in the SAME block_timestamp; only logIndex distinguishes order. + # Sorting by round_id (the old bug) would reorder these into 101->102 and hide + # the regression. Sorting by (blockNumber, logIndex) preserves the real stream. + rounds = [ + _round(102, 100_000_000, 1_000, block_number=500, log_index=0), + _round(101, 100_000_000, 1_000, block_number=500, log_index=1), + ] + alerts = _detect(rounds) + self.assertTrue(any(a.severity == AlertSeverity.CRITICAL for a in alerts)) + self.assertTrue(any("sequence anomaly" in a.message for a in alerts)) + class TestDedup(unittest.TestCase): def test_rounds_at_or_before_cursor_not_realerted(self): @@ -126,6 +149,8 @@ def test_maps_graphql_row(self): "current": "99980000", "updatedAt": "1700000000", "blockTimestamp": "1700000005", + "blockNumber": "21000000", + "logIndex": "7", "transactionHash": "0xdead", "chainId": 1, } @@ -135,6 +160,7 @@ def test_maps_graphql_row(self): self.assertEqual(rnd.answer, 99_980_000) self.assertEqual(rnd.updated_at, 1_700_000_000) self.assertEqual(rnd.block_timestamp, 1_700_000_005) + self.assertEqual(rnd.event_order, (21_000_000, 7)) if __name__ == "__main__": From 2760697fdffc562cdc2cbd88889527887b38a978 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 30 Jun 2026 12:54:11 +0200 Subject: [PATCH 06/10] refactor(pegs): drop stale/round-sanity helpers, track iUSD over siUSD - Remove is_stale and round_issues/is_round_healthy from utils/chainlink.py: no longer reliable oracle health indicators and unused by production code. - Replace the unverified siUSD placeholder entry with the existing verified iUSD (infinifi) entry; siUSD is yield-bearing staked iUSD and shouldn't be modeled as a flat-peg deviation. Drop the now-unused PLACEHOLDER_ADDRESS. - Update tests to match. Co-Authored-By: Claude Opus 4.8 --- tests/test_chainlink.py | 40 ---------------------------- tests/test_pegged_assets.py | 2 +- utils/chainlink.py | 53 ++----------------------------------- utils/pegged_assets.py | 14 +--------- 4 files changed, 4 insertions(+), 105 deletions(-) diff --git a/tests/test_chainlink.py b/tests/test_chainlink.py index 032594c0..f174dfc5 100644 --- a/tests/test_chainlink.py +++ b/tests/test_chainlink.py @@ -4,9 +4,6 @@ from utils.chainlink import ( FeedReading, RoundData, - is_round_healthy, - is_stale, - round_issues, scale_price, ) @@ -36,43 +33,6 @@ def test_negative_decimals_raises(self): scale_price(1, -1) -class TestIsStale(unittest.TestCase): - def test_fresh_within_heartbeat(self): - self.assertFalse(is_stale(updated_at=1_000, heartbeat=3_600, now=4_000)) - - def test_stale_past_heartbeat(self): - self.assertTrue(is_stale(updated_at=1_000, heartbeat=3_600, now=5_000)) - - def test_buffer_extends_window(self): - # 4000s elapsed, 3600 heartbeat -> stale, but a 600s buffer keeps it fresh. - self.assertFalse(is_stale(updated_at=1_000, heartbeat=3_600, now=5_000, buffer=600)) - - def test_exactly_at_heartbeat_is_not_stale(self): - self.assertFalse(is_stale(updated_at=1_000, heartbeat=3_600, now=4_600)) - - def test_uninitialised_updated_at_is_stale(self): - self.assertTrue(is_stale(updated_at=0, heartbeat=3_600, now=4_000)) - - -class TestRoundSanity(unittest.TestCase): - def test_healthy_round_has_no_issues(self): - self.assertEqual(round_issues(_round()), []) - self.assertTrue(is_round_healthy(_round())) - - def test_non_positive_answer(self): - issues = round_issues(_round(answer=0)) - self.assertTrue(any("non-positive answer" in i for i in issues)) - self.assertFalse(is_round_healthy(_round(answer=-5))) - - def test_incomplete_round(self): - issues = round_issues(_round(updated_at=0)) - self.assertTrue(any("not complete" in i for i in issues)) - - def test_stale_answered_in_round(self): - issues = round_issues(_round(round_id=10, answered_in_round=9)) - self.assertTrue(any("stale round" in i for i in issues)) - - class TestRoundData(unittest.TestCase): def test_from_tuple_decodes_fields(self): rd = RoundData.from_tuple((10, 99_851_375, 900, 1_000, 10)) diff --git a/tests/test_pegged_assets.py b/tests/test_pegged_assets.py index a15de9ff..3a20f5a3 100644 --- a/tests/test_pegged_assets.py +++ b/tests/test_pegged_assets.py @@ -13,7 +13,7 @@ ) # Asset set the registry must cover per the issue acceptance criteria. -REQUIRED_ASSETS = {"cbBTC", "LBTC", "siUSD", "cUSD", "USDe", "USDC", "USDT", "USDS"} +REQUIRED_ASSETS = {"cbBTC", "LBTC", "iUSD", "cUSD", "USDe", "USDC", "USDT", "USDS"} class TestPriceDeviation(unittest.TestCase): diff --git a/utils/chainlink.py b/utils/chainlink.py index 79b50bf1..ea65181b 100644 --- a/utils/chainlink.py +++ b/utils/chainlink.py @@ -1,9 +1,8 @@ """Chainlink aggregator helpers shared across peg / oracle monitors. Generalises the inline ``latestRoundData`` handling from ``protocols/ustb/main.py``: -a batched feed reader plus pure helpers for staleness, round sanity checks and -price scaling. Pure helpers take primitive values so they are trivially unit -testable without a chain connection. +a batched feed reader plus a pure price-scaling helper that takes primitive +values so it is trivially unit testable without a chain connection. """ from dataclasses import dataclass @@ -87,54 +86,6 @@ def scale_price(answer: int, decimals: int) -> Decimal: return Decimal(answer) / (Decimal(10) ** decimals) -def is_stale(updated_at: int, heartbeat: int, now: int, buffer: int = 0) -> bool: - """Return ``True`` if a feed has not updated within its heartbeat window. - - Args: - updated_at: ``updatedAt`` timestamp from the latest round (unix seconds). - heartbeat: Expected maximum interval between updates (seconds). - now: Current unix timestamp (seconds). - buffer: Extra grace period added to the heartbeat before flagging - staleness (seconds). Defaults to ``0``. - - Returns: - ``True`` when ``now - updated_at`` exceeds ``heartbeat + buffer``, or when - ``updated_at`` is non-positive (uninitialised / invalid round). - """ - if updated_at <= 0: - return True - return (now - updated_at) > (heartbeat + buffer) - - -def round_issues(round_data: RoundData) -> list[str]: - """Collect Chainlink round sanity-check failures. - - Checks the standard freshness/consistency invariants Chainlink consumers are - expected to enforce: a positive answer, an initialised round, and an - ``answeredInRound`` that is not behind the current ``roundId`` (a stale answer - carried over from an earlier round). - - Args: - round_data: The decoded round data to validate. - - Returns: - A list of human-readable problem descriptions; empty when healthy. - """ - issues: list[str] = [] - if round_data.answer <= 0: - issues.append(f"non-positive answer ({round_data.answer})") - if round_data.updated_at <= 0: - issues.append("round not complete (updatedAt is 0)") - if round_data.answered_in_round < round_data.round_id: - issues.append(f"stale round (answeredInRound {round_data.answered_in_round} < roundId {round_data.round_id})") - return issues - - -def is_round_healthy(round_data: RoundData) -> bool: - """Return ``True`` if the round passes all sanity checks in :func:`round_issues`.""" - return not round_issues(round_data) - - # --------------------------------------------------------------------------- # Batched on-chain reader # --------------------------------------------------------------------------- diff --git a/utils/pegged_assets.py b/utils/pegged_assets.py index a5b77424..7cab2247 100644 --- a/utils/pegged_assets.py +++ b/utils/pegged_assets.py @@ -11,8 +11,7 @@ entry covers both dollar- and bitcoin-denominated assets. ``depeg_pct`` is a *deviation* tolerance (fractional distance from the peg), not an absolute floor. -Addresses and Chainlink feeds were verified on Ethereum mainnet; entries that -could not be verified use :data:`PLACEHOLDER_ADDRESS` and are marked ``TODO``. +Addresses and Chainlink feeds were verified on Ethereum mainnet. """ from dataclasses import dataclass @@ -24,9 +23,6 @@ # DeFiLlama key for the live BTC/USD reference price (BTC peg target). BTC_USD_DEFILLAMA_KEY = "coingecko:bitcoin" -# Sentinel for assets/feeds whose address has not yet been verified on-chain. -PLACEHOLDER_ADDRESS = "0x0000000000000000000000000000000000000000" - class PegTarget(Enum): """What an asset is pegged to. @@ -209,14 +205,6 @@ def get_asset(name: str) -> PeggedAsset: peg=PegTarget.USD, depeg_pct=Decimal("0.03"), ), - PeggedAsset( - # TODO: siUSD token address not yet verified on-chain — placeholder. - name="siUSD", - defillama_key=f"ethereum:{PLACEHOLDER_ADDRESS}", - channel="pegs", - peg=PegTarget.USD, - depeg_pct=Decimal("0.05"), - ), # --- BTC-pegged ----------------------------------------------------------- PeggedAsset( name="cbBTC", From 6aa3c13d984991b64505dd455cc837b56cc85ea8 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 30 Jun 2026 13:34:02 +0200 Subject: [PATCH 07/10] feat(pegs): WBTC peg, BTC-wrapper downside-only depeg, wstETH pool monitor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add WBTC to the peg registry (BTC peg, verified WBTC/BTC Chainlink feed). - ChainlinkFeed gains an explicit `quote: PegTarget` so consumers can interpret mixed feed denominations (WBTC/LBTC are BTC-quoted ~1.0; cbBTC is USD-quoted). - PeggedAsset gains `downside_only`; is_depegged becomes asymmetric for BTC wrappers (WBTC/cbBTC/LBTC) so a legit move above peg (e.g. LBTC at ~1.004 BTC) no longer false-alerts — only a drop below peg triggers. - lrt-pegs curve monitor: read pools via balances(i) (works for modern and legacy pools) and add the deep Lido stETH/ETH pool as the canonical wstETH-vs-ETH depeg gauge (wstETH deterministically wraps stETH). Co-Authored-By: Claude Opus 4.8 --- protocols/lrt-pegs/README.md | 2 +- protocols/lrt-pegs/curve/main.py | 26 +++++++++------- tests/test_pegged_assets.py | 29 +++++++++++++++++- utils/pegged_assets.py | 51 +++++++++++++++++++++++++++----- 4 files changed, 89 insertions(+), 19 deletions(-) diff --git a/protocols/lrt-pegs/README.md b/protocols/lrt-pegs/README.md index 65449453..f7748b49 100644 --- a/protocols/lrt-pegs/README.md +++ b/protocols/lrt-pegs/README.md @@ -10,8 +10,8 @@ Curve pools that are checked are: - ETH+/WETH - weETH-WETH -- frxETH-WETH - OETH/ETH +- stETH/ETH (Lido) — canonical wstETH-vs-ETH depeg gauge; wstETH deterministically wraps stETH ### Uniswap V3 pools - DISABLED ⚠️ diff --git a/protocols/lrt-pegs/curve/main.py b/protocols/lrt-pegs/curve/main.py index 0f5d5070..1a817c85 100644 --- a/protocols/lrt-pegs/curve/main.py +++ b/protocols/lrt-pegs/curve/main.py @@ -18,28 +18,34 @@ ("OETH/ETH Curve Pool", "0xcc7d5785AD5755B6164e21495E07aDb0Ff11C2A8", 0, 1, THRESHOLD_RATIO, "origin"), # NOTE: bool is unbalanced, whole liquidity is moved to univ3: https://app.uniswap.org/explore/pools/ethereum/0x202a6012894ae5c288ea824cbc8a9bfb26a49b93 ("weETH-WETH Curve Pool", "0xDB74dfDD3BB46bE8Ce6C33dC9D82777BCFc3dEd5", 1, 0, THRESHOLD_RATIO, "weeth"), + # Lido stETH/ETH — deepest stETH<>ETH venue and the canonical wstETH depeg + # gauge (wstETH deterministically wraps stETH). Legacy pool: exposes + # balances(i) but not get_balances(). idx 0 = ETH, idx 1 = stETH. + ("stETH/ETH Curve Pool", "0xDC24316b9AE028F1497c275EB9192a3Ea0f67022", 1, 0, THRESHOLD_RATIO, "wsteth"), ] def process_pools(chain: Chain = Chain.MAINNET): client = ChainManager.get_client(chain) - contracts = [] - # Prepare batch calls + # Read each pool's two relevant coin balances. Using ``balances(i)`` (instead + # of ``get_balances()``) keeps a single code path for both modern pools and + # legacy pools like Lido stETH/ETH that don't expose ``get_balances()``. with client.batch_requests() as batch: - for _, pool_address, _, _, _, _ in POOL_CONFIGS: + for _, pool_address, idx_lrt, idx_other_token, _, _ in POOL_CONFIGS: pool = client.eth.contract(address=pool_address, abi=ABI_CURVE_POOL) - contracts.append(pool) - - batch.add(pool.functions.get_balances()) + batch.add(pool.functions.balances(idx_lrt)) + batch.add(pool.functions.balances(idx_other_token)) responses = client.execute_batch(batch) - if len(responses) != len(POOL_CONFIGS): - raise ValueError(f"Expected {len(POOL_CONFIGS)} responses from batch, got: {len(responses)}") + if len(responses) != len(POOL_CONFIGS) * 2: + raise ValueError(f"Expected {len(POOL_CONFIGS) * 2} responses from batch, got: {len(responses)}") # Process results - for (pool_name, _, idx_lrt, idx_other_token, peg_threshold, protocol), balances in zip(POOL_CONFIGS, responses): - percentage = (balances[idx_lrt] / (balances[idx_lrt] + balances[idx_other_token])) * 100 + for i, (pool_name, _, _, _, peg_threshold, protocol) in enumerate(POOL_CONFIGS): + lrt_balance = responses[i * 2] + other_balance = responses[i * 2 + 1] + percentage = (lrt_balance / (lrt_balance + other_balance)) * 100 logger.info("%s ratio is %s%%", pool_name, f"{percentage:.2f}") if percentage > peg_threshold: message = f"🚨 Curve Alert! {pool_name} ratio is {percentage:.2f}%" diff --git a/tests/test_pegged_assets.py b/tests/test_pegged_assets.py index 3a20f5a3..18eb7fc8 100644 --- a/tests/test_pegged_assets.py +++ b/tests/test_pegged_assets.py @@ -13,7 +13,7 @@ ) # Asset set the registry must cover per the issue acceptance criteria. -REQUIRED_ASSETS = {"cbBTC", "LBTC", "iUSD", "cUSD", "USDe", "USDC", "USDT", "USDS"} +REQUIRED_ASSETS = {"WBTC", "cbBTC", "LBTC", "iUSD", "cUSD", "USDe", "USDC", "USDT", "USDS"} class TestPriceDeviation(unittest.TestCase): @@ -53,6 +53,19 @@ def test_btc_asset_uses_peg_price(self): self.assertFalse(cbbtc.is_depegged(Decimal("60500"), Decimal("60000"))) self.assertTrue(cbbtc.is_depegged(Decimal("58000"), Decimal("60000"))) + def test_downside_only_ignores_upside(self): + lbtc = get_asset("LBTC") # depeg_pct = 0.03, downside_only + # +5% above peg (LBTC legitimately trades above 1 BTC) -> not a depeg. + self.assertFalse(lbtc.is_depegged(Decimal("63000"), Decimal("60000"))) + # -5% below peg -> depeg. + self.assertTrue(lbtc.is_depegged(Decimal("57000"), Decimal("60000"))) + # exactly at the downside threshold -> depeg. + self.assertTrue(lbtc.is_depegged(Decimal("58200"), Decimal("60000"))) + + def test_symmetric_asset_flags_upside(self): + usdc = get_asset("USDC") # depeg_pct = 0.02, symmetric + self.assertTrue(usdc.is_depegged(Decimal("1.05"), Decimal("1"))) + class TestResolvePegPrices(unittest.TestCase): def test_usd_only_does_not_hit_network(self): @@ -96,6 +109,20 @@ def test_every_asset_has_positive_depeg_tolerance(self): for asset in PEGGED_ASSETS: self.assertGreater(asset.depeg_pct, Decimal("0"), asset.name) + def test_btc_wrappers_are_downside_only(self): + for name in ("WBTC", "cbBTC", "LBTC"): + self.assertTrue(get_asset(name).downside_only, name) + + def test_usd_stables_are_symmetric(self): + self.assertFalse(get_asset("USDC").downside_only) + + def test_chainlink_feed_quote_denomination(self): + # BTC-denominated feeds (answer ~1.0) vs USD-denominated feeds (absolute price). + self.assertEqual(get_asset("WBTC").chainlink_feed.quote, PegTarget.BTC) + self.assertEqual(get_asset("LBTC").chainlink_feed.quote, PegTarget.BTC) + self.assertEqual(get_asset("cbBTC").chainlink_feed.quote, PegTarget.USD) + self.assertEqual(get_asset("USDC").chainlink_feed.quote, PegTarget.USD) + if __name__ == "__main__": unittest.main() diff --git a/utils/pegged_assets.py b/utils/pegged_assets.py index 7cab2247..b667f29c 100644 --- a/utils/pegged_assets.py +++ b/utils/pegged_assets.py @@ -3,7 +3,7 @@ This registry is consumed by every layer of peg/oracle monitoring: * L1 — market depeg (DeFiLlama price vs ``peg`` target, deviation > ``depeg_pct``) -* L2 — oracle health (``chainlink_feed`` staleness / round sanity, ``rate_oracle`` drift) +* L2 — oracle health (``chainlink_feed`` price cross-check, ``rate_oracle`` drift) * L3 — event consumers Peg deviation is expressed relative to a :class:`PegTarget` (``USD`` is the @@ -37,11 +37,22 @@ class PegTarget(Enum): @dataclass(frozen=True) class ChainlinkFeed: - """A Chainlink aggregator backing an asset's price (consumed by L2).""" + """A Chainlink aggregator backing an asset's price (consumed by L2). + + Args: + address: Aggregator contract address. + heartbeat: Max expected seconds between updates (Chainlink mainnet default). + description: Human-readable feed pair, e.g. ``"WBTC/BTC"``. + quote: Denomination of the feed's ``answer``. A ``USD`` feed reports an + absolute price; a ``BTC`` feed reports the asset-to-BTC ratio (~1.0). + Lets consumers scale correctly — BTC-quoted feeds compare straight to + ``1.0`` with no BTC/USD lookup, USD-quoted feeds need live BTC/USD. + """ address: str - heartbeat: int # max expected seconds between updates (Chainlink mainnet default) + heartbeat: int description: str = "" + quote: PegTarget = PegTarget.USD @dataclass(frozen=True) @@ -74,6 +85,9 @@ class PeggedAsset: depeg_pct: Decimal # deviation tolerance from the peg (e.g. Decimal("0.02") = 2%) chainlink_feed: ChainlinkFeed | None = None rate_oracle: RateOracle | None = None + # When True, only a drop *below* the peg counts as a depeg; upside is ignored. + # Use for assets that can legitimately trade above peg (e.g. BTC wrappers). + downside_only: bool = False @property def address(self) -> str: @@ -85,8 +99,16 @@ def deviation(self, price: Decimal, peg_price: Decimal) -> Decimal: return price_deviation(price, peg_price) def is_depegged(self, price: Decimal, peg_price: Decimal) -> bool: - """Return ``True`` if ``price`` deviates from ``peg_price`` beyond ``depeg_pct``.""" - return abs(self.deviation(price, peg_price)) >= self.depeg_pct + """Return ``True`` if ``price`` has depegged from ``peg_price`` beyond ``depeg_pct``. + + For ``downside_only`` assets only a drop below the peg triggers (deviation + ``<= -depeg_pct``); for all others the check is symmetric (``abs`` deviation + ``>= depeg_pct``), so an upside move flags too. + """ + deviation = self.deviation(price, peg_price) + if self.downside_only: + return deviation <= -self.depeg_pct + return abs(deviation) >= self.depeg_pct # --------------------------------------------------------------------------- @@ -206,6 +228,17 @@ def get_asset(name: str) -> PeggedAsset: depeg_pct=Decimal("0.03"), ), # --- BTC-pegged ----------------------------------------------------------- + PeggedAsset( + name="WBTC", + defillama_key="ethereum:0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599", + channel="pegs", + peg=PegTarget.BTC, + depeg_pct=Decimal("0.02"), + chainlink_feed=ChainlinkFeed( + "0xfdFD9C85aD200c506Cf9e21F1FD8dD01932FBB23", _STABLE_HEARTBEAT, "WBTC/BTC", quote=PegTarget.BTC + ), + downside_only=True, # only a drop below BTC is a risk + ), PeggedAsset( name="cbBTC", defillama_key="ethereum:0xcbB7C0000aB88B473b1f5aFd9ef808440eed33Bf", @@ -213,6 +246,7 @@ def get_asset(name: str) -> PeggedAsset: peg=PegTarget.BTC, depeg_pct=Decimal("0.02"), chainlink_feed=ChainlinkFeed("0x2665701293fCbEB223D11A08D826563EDcCE423A", _STABLE_HEARTBEAT, "cbBTC/USD"), + downside_only=True, # only a drop below BTC is a risk ), PeggedAsset( name="LBTC", @@ -220,8 +254,11 @@ def get_asset(name: str) -> PeggedAsset: channel="pegs", peg=PegTarget.BTC, depeg_pct=Decimal("0.03"), - # LBTC/BTC market-rate feed (8 decimals); price ~1 BTC per LBTC. - chainlink_feed=ChainlinkFeed("0x5c29868C58b6e15e2b962943278969Ab6a7D3212", _STABLE_HEARTBEAT, "LBTC/BTC"), + # LBTC/BTC market-rate feed (8 decimals); can sit slightly above 1 BTC. + chainlink_feed=ChainlinkFeed( + "0x5c29868C58b6e15e2b962943278969Ab6a7D3212", _STABLE_HEARTBEAT, "LBTC/BTC", quote=PegTarget.BTC + ), + downside_only=True, # LBTC can trade above peg; only a drop below BTC matters ), ] From ad3ca605b8b8cffbbe3137455e28d1b3b9d71317 Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 30 Jun 2026 13:43:21 +0200 Subject: [PATCH 08/10] chore: zero decimal error --- tests/test_chainlink.py | 5 +++-- utils/chainlink.py | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/test_chainlink.py b/tests/test_chainlink.py index f174dfc5..348b7836 100644 --- a/tests/test_chainlink.py +++ b/tests/test_chainlink.py @@ -25,8 +25,9 @@ def test_scales_by_decimals(self): def test_scales_fractional(self): self.assertEqual(scale_price(99_960_043, 8), Decimal("0.99960043")) - def test_zero_decimals_is_identity(self): - self.assertEqual(scale_price(42, 0), Decimal("42")) + def test_zero_decimals_raises(self): + with self.assertRaises(ValueError): + scale_price(1, 0) def test_negative_decimals_raises(self): with self.assertRaises(ValueError): diff --git a/utils/chainlink.py b/utils/chainlink.py index ea65181b..93d3adb9 100644 --- a/utils/chainlink.py +++ b/utils/chainlink.py @@ -81,8 +81,8 @@ def scale_price(answer: int, decimals: int) -> Decimal: Raises: ValueError: If ``decimals`` is negative. """ - if decimals < 0: - raise ValueError(f"decimals must be non-negative, got {decimals}") + if decimals < 1: + raise ValueError(f"decimals must be positive, got {decimals}") return Decimal(answer) / (Decimal(10) ** decimals) From 477f4c86374df5406bf1c7687b1a67c5a5aaa26b Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 30 Jun 2026 17:51:31 +0200 Subject: [PATCH 09/10] chore: remove lbtc info --- protocols/stables/oracles.py | 21 ++------------------- utils/dispatch.py | 17 ++++++++++++++++- 2 files changed, 18 insertions(+), 20 deletions(-) diff --git a/protocols/stables/oracles.py b/protocols/stables/oracles.py index 319564e4..f58dea11 100644 --- a/protocols/stables/oracles.py +++ b/protocols/stables/oracles.py @@ -17,9 +17,7 @@ For **rate / fundamental oracles** (vault-rate, capped Redstone feeds) it checks monotonicity + delta-vs-cached (the ``protocols/apyusd/main.py`` approach); any -fundamental-oracle depeg is ``CRITICAL`` (per #196). Fundamental oracles already -covered by a Tenderly alert are listed in :data:`TENDERLY_COVERED` and skipped -here to avoid duplicate alerting. +fundamental-oracle depeg is ``CRITICAL`` (per #196). Runs hourly via ``automation/jobs.yaml``. """ @@ -56,21 +54,6 @@ def _rate_cache_key(address: str) -> str: return f"peg_oracle_rate_{address.lower()}" -# Fundamental oracles already covered by an existing Tenderly alert — NOT polled -# here to avoid duplicate alerting. See protocols/lrt-pegs/README.md. -# -# Gap analysis (per #196 step 6): the registry currently exposes no *uncovered* -# fundamental oracle. Adding active polling for a new one needs the oracle -# contract address, its read function + precision, and whether it is monotonic -# (capped) — wire it as a ``rate_oracle`` on the relevant ``PeggedAsset``. -TENDERLY_COVERED: dict[str, str] = { - # LBTC Redstone fundamental oracle, upper-capped at 1 (healthy == 1). - "LBTC Redstone (0xb415eAA355D8440ac7eCB602D3fb67ccC1f0bc81)": ( - "https://dashboard.tenderly.co/yearn/sam/alerts/rules/eca272ef-979a-47b3-a7f0-2e67172889bb" - ), -} - - # --------------------------------------------------------------------------- # Observation + pure per-check helpers (unit tested without a chain) # --------------------------------------------------------------------------- @@ -388,7 +371,7 @@ def main() -> None: client = ChainManager.get_client(Chain.MAINNET) _monitor_chainlink_assets(client) _monitor_rate_oracles(client) - logger.info("L2 oracle health check complete (%d Tenderly-covered oracle(s) skipped)", len(TENDERLY_COVERED)) + logger.info("L2 oracle health check complete") if __name__ == "__main__": diff --git a/utils/dispatch.py b/utils/dispatch.py index e68680aa..72930f7b 100644 --- a/utils/dispatch.py +++ b/utils/dispatch.py @@ -29,7 +29,22 @@ # Protocols that have emergency withdrawal config in liquidity-monitoring. # Only these protocols will trigger a dispatch. -DISPATCHABLE_PROTOCOLS = {"infinifi", "cap", "ethena", "ethplus", "usdai", "origin", "maple", "3jane", "wbtc", "coinbase", "lombard", "tether", "circle", "maker"} +DISPATCHABLE_PROTOCOLS = { + "infinifi", + "cap", + "ethena", + "ethplus", + "usdai", + "origin", + "maple", + "3jane", + "wbtc", + "coinbase", + "lombard", + "tether", + "circle", + "maker", +} def _is_on_cooldown(protocol: str, cooldown_seconds: int = DEFAULT_COOLDOWN_SECONDS) -> bool: From a60af2030b804843f1515eeaa757f8b9f458f36b Mon Sep 17 00:00:00 2001 From: spalen0 Date: Tue, 30 Jun 2026 17:59:46 +0200 Subject: [PATCH 10/10] feat(pegs): gate L3 event sequence/gap checks on reports_round_metadata The sequence-anomaly (roundId) and missed-heartbeat-gap (updatedAt) checks in oracle_events.py rely on a feed reporting reliable round metadata. Honor the same ChainlinkFeed.reports_round_metadata flag L2 uses so feeds that return constant or zero roundId/updatedAt don't false-positive; the answer-based jump check always runs. Also warn when an AnswerUpdated query hits its row limit (newest rounds deferred to the next run). Adds gating tests. Co-Authored-By: Claude Opus 4.8 --- protocols/stables/oracle_events.py | 23 ++++++++++++++---- tests/test_stables_oracle_events.py | 37 ++++++++++++++++++++++++++++- 2 files changed, 55 insertions(+), 5 deletions(-) diff --git a/protocols/stables/oracle_events.py b/protocols/stables/oracle_events.py index 6a40b133..ccc8f259 100644 --- a/protocols/stables/oracle_events.py +++ b/protocols/stables/oracle_events.py @@ -150,6 +150,11 @@ def detect_anomalies( ``rounds`` may include up to one round at/just before ``since_ts`` for context (jump / gap diffing); alerts only fire for the rounds whose ``block_timestamp`` is strictly greater than ``since_ts`` so reruns never re-alert the same round. + + The sequence and missed-heartbeat checks rely on a meaningful ``roundId`` / + ``updatedAt``; for feeds flagged ``reports_round_metadata=False`` (constant or + zero round/time values) they are skipped and only the answer-based jump check + runs — mirroring L2's gating in ``oracles.py``. """ # Sort by the true emission order (blockNumber, logIndex), NOT round_id — see # OracleRound.event_order. Using round_id would mask a backwards round whenever @@ -161,8 +166,9 @@ def detect_anomalies( if cur.block_timestamp <= since_ts: continue # already processed in a prior run - # Sequence anomaly: roundId must strictly increase. - if cur.round_id <= prev.round_id: + # Sequence anomaly: roundId must strictly increase. Skipped for feeds that + # don't report a meaningful roundId (constant/zero) to avoid false positives. + if feed.reports_round_metadata and cur.round_id <= prev.round_id: alerts.append( _alert( asset, @@ -190,9 +196,10 @@ def detect_anomalies( ) ) - # Missed-heartbeat gap between consecutive updates. + # Missed-heartbeat gap between consecutive updates. Relies on a meaningful + # updatedAt, so skipped when the feed doesn't report reliable round metadata. gap = cur.updated_at - prev.updated_at - if gap > feed.heartbeat + heartbeat_buffer: + if feed.reports_round_metadata and gap > feed.heartbeat + heartbeat_buffer: alerts.append( _alert( asset, @@ -401,6 +408,14 @@ def main() -> None: rows = response.get("data", {}).get("AnswerUpdated", []) _logger.info("Fetched %d AnswerUpdated rows", len(rows)) + if len(rows) >= args.limit: + # Rows are ordered ascending, so the newest rounds are the ones dropped; the + # cursor still advances to the last fetched round, so they land next run. + _logger.warning( + "AnswerUpdated hit the query limit (%d) — newest rounds deferred to the next run; " + "raise --limit / PEG_EVENT_QUERY_LIMIT if this recurs", + args.limit, + ) process_rounds(aggregator_map, rows, use_cache) diff --git a/tests/test_stables_oracle_events.py b/tests/test_stables_oracle_events.py index e5f383a7..3f7e8d42 100644 --- a/tests/test_stables_oracle_events.py +++ b/tests/test_stables_oracle_events.py @@ -1,4 +1,5 @@ import unittest +from decimal import Decimal from protocols.stables.oracle_events import ( OracleRound, @@ -8,7 +9,7 @@ ) from utils.alert import AlertSeverity from utils.dispatch import DISPATCHABLE_PROTOCOLS -from utils.pegged_assets import get_asset +from utils.pegged_assets import ChainlinkFeed, PeggedAsset, PegTarget, get_asset USDE = get_asset("USDe") # protocol "ethena" (dispatchable), USD/USD feed, 24h heartbeat FEED = USDE.chainlink_feed @@ -102,6 +103,40 @@ def test_backwards_round_in_same_block_is_detected(self): self.assertTrue(any("sequence anomaly" in a.message for a in alerts)) +class TestRoundMetadataGate(unittest.TestCase): + """Feeds flagged reports_round_metadata=False skip sequence + gap, keep jump.""" + + def _flat_asset(self) -> PeggedAsset: + return PeggedAsset( + name="fakeFlat", + defillama_key="ethereum:0x0000000000000000000000000000000000000002", + protocol="pegs", + peg=PegTarget.USD, + depeg_pct=Decimal("0.02"), + chainlink_feed=ChainlinkFeed("0xfeed", HB, "FLAT/USD", reports_round_metadata=False), + ) + + def _detect_flat(self, rounds): + asset = self._flat_asset() + return detect_anomalies(asset, asset.chainlink_feed, rounds, since_ts=0) + + def test_sequence_and_gap_skipped_when_metadata_unreliable(self): + # Non-increasing roundId AND a heartbeat-busting gap; both would normally fire, + # but the feed is flagged unreliable so neither does (answer barely moves). + rounds = [ + _round(101, 100_000_000, 1_000), + _round(101, 100_000_001, 1_000 + HB + 50_000), + ] + self.assertEqual(self._detect_flat(rounds), []) + + def test_jump_still_fires_when_metadata_unreliable(self): + # The answer-based jump check is independent of round metadata. + rounds = [_round(100, 100_000_000, 1_000), _round(101, 130_000_000, 1_500)] # +30% + alerts = self._detect_flat(rounds) + self.assertEqual(len(alerts), 1) + self.assertIn("jump", alerts[0].message) + + class TestDedup(unittest.TestCase): def test_rounds_at_or_before_cursor_not_realerted(self): # The jump happens between round 100->101; both at/below the cursor.