feat(pegs): L3 oracle-events consumer — stables/oracle_events.py (#302)

automation/jobs.yaml

@@ -37,6 +37,8 @@ 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: "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 }

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 ⚠️
 

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}%"

protocols/morpho/_shared.py

@@ -13,12 +13,14 @@
 logger = get_logger("morpho.shared")
 
 
-# Yearn-curated Morpho V2 vaults — sourced from
+# Morpho V2 vaults we monitor. Mostly Yearn-curated — sourced from
 # https://app.morpho.org/curator/yearn?v2=true (filtered via GraphQL by
-# Yearn's curator addresses). Imported by both ``markets_v2.py`` and
-# ``governance_v2.py``. To add a new vault, append a
-# ``[name, address, risk_level]`` row to the appropriate chain. Risk levels
-# follow the same 1–5 scheme as v1 ``markets.py:VAULTS_BY_CHAIN``.
+# Yearn's curator addresses) — plus a few third-party vaults (Gauntlet,
+# Steakhouse) the team tracks because of Yearn's exposure to them.
+# Imported by both ``markets_v2.py`` and ``governance_v2.py``, so adding a row
+# here enrols the vault in market checks *and* governance diffs. To add a new
+# vault, append a ``[name, address, risk_level]`` row to the appropriate chain.
+# Risk levels follow the same 1–5 scheme as v1 ``markets.py:VAULTS_BY_CHAIN``.
 VAULTS_V2_BY_CHAIN: Dict[Chain, List[List[Any]]] = {
     Chain.MAINNET: [
         # name, address, risk level
@@ -44,6 +46,11 @@
         # ["Yearn OG ETH", "0x5920A6FC553af799542EDA628AdfCc9eA52e141C", 1],
         ["Yearn KAT", "0x9b1aE9548E4B46cEB6650f6CEc702bAf5CF2b8CC", 1],
         ["Yearn Degen USDC", "0xA2d38c8A3D810EBcF4C2075821c5eC8F976bb692", 3],
+        # Synced with /srv/monitoring config (KATANA V2 vaults the team already
+        # tracks in prod). Curated by Gauntlet and Steakhouse, not Yearn.
+        ["Gauntlet USDT", "0xaC596AD9771a8d0D4DF108ae0406e6f913aEdceb", 1],
+        ["Steakhouse High Yield USDC", "0xbeeff2d5d126d4809195EeA02b605423917bb6c6", 2],
+        ["Steakhouse Prime USDC", "0xbeef042bAD4472c3F7Eb9A73070703788b5362D7", 1],
     ],
 }
 

protocols/morpho/governance_v2.py

@@ -287,27 +287,122 @@ def _pending_function_key(snapshot: V2GovernanceSnapshot, data_hash: str) -> str
     return morpho_key(snapshot.address.lower(), data_hash, PENDING_FUNCTION_TYPE)
 
 
-def _alert_pending_new(snapshot: V2GovernanceSnapshot, pc: PendingConfig, operation_label: str) -> None:
-    send_alert(
-        Alert(
-            AlertSeverity.MEDIUM,
-            f"⏳ V2 [{snapshot.name}]({get_vault_url(snapshot.address, snapshot.chain)}) "
-            f"on {snapshot.chain.name}\n"
-            f"📥 Submitted: {operation_label}\n"
-            f"⏰ Executable at: {_format_ts(pc.valid_at)} {_format_countdown(pc.valid_at)}\n"
-            f"🔗 Tx: {_explorer_link(snapshot.chain, pc.tx_hash)}",
-            PROTOCOL,
-        )
+@dataclass
+class _VaultAlert:
+    """One section of a vault's grouped Telegram message."""
+
+    severity: AlertSeverity
+    body: str
+
+
+@dataclass
+class _VaultDiff:
+    """Buffered output of one vault's diff pass: alert sections and cache writes.
+
+    Each diff category (``_diff_pending``, ``_diff_single_role``, ``_diff_set``)
+    appends its sections here instead of sending immediately, so a vault with 3
+    new pending configs, 1 owner change, and 1 adapter swap arrives as one alert
+    instead of 5.
+
+    Cache writes are buffered too, and committed only once the Telegram send has
+    succeeded (see ``diff_and_alert``). Writing them during the diff pass would
+    mark changes as alerted even when delivery failed — ``main`` swallows the
+    exception, so the alert would be lost for good.
+    """
+
+    alerts: List[_VaultAlert] = field(default_factory=list)
+    writes: List[tuple[str, Any]] = field(default_factory=list)
+
+    def alert(self, severity: AlertSeverity, body: str) -> None:
+        """Buffer one section of the vault's grouped message."""
+        self.alerts.append(_VaultAlert(severity, body))
+
+    def write(self, key: str, value: Any) -> None:
+        """Buffer a cache write to apply after the alert is delivered."""
+        self.writes.append((key, value))
+
+    def commit(self) -> None:
+        """Persist every buffered cache write."""
+        for key, value in self.writes:
+            _write(key, value)
+
+
+# Ascending severity order — a grouped alert takes the highest of its sections.
+_SEVERITY_ORDER = (AlertSeverity.LOW, AlertSeverity.MEDIUM, AlertSeverity.HIGH, AlertSeverity.CRITICAL)
+
+# Telegram caps a message at 4096 chars, and ``utils.telegram`` truncates
+# anything longer (dropping Markdown with it), so a large batch would silently
+# lose its tail. We split into "(i/N)" parts instead. The budget leaves headroom
+# for the header line, the part suffix, and the emoji ``send_alert`` prepends.
+_MAX_MESSAGE_CHARS = 3900
+_SECTION_SEPARATOR = "\n\n---\n\n"
+
+
+def _vault_header(snapshot: V2GovernanceSnapshot) -> str:
+    """One-line header for the grouped alert: ``V2 [name](url) on chain``."""
+    return f"V2 [{snapshot.name}]({get_vault_url(snapshot.address, snapshot.chain)}) on {snapshot.chain.name}"
+
+
+def _split_into_messages(alerts: List[_VaultAlert], budget: int) -> List[List[str]]:
+    """Pack section bodies into groups that each fit within ``budget`` chars.
+
+    A single section bigger than the budget still gets its own message rather
+    than pushing its neighbours out: Telegram truncates that one section, but no
+    other section is lost.
+    """
+    parts: List[List[str]] = [[]]
+    size = 0
+    for alert in alerts:
+        cost = len(alert.body) + len(_SECTION_SEPARATOR)
+        if parts[-1] and size + cost > budget:
+            parts.append([])
+            size = 0
+        parts[-1].append(alert.body)
+        size += cost
+    return parts
+
+
+def _send_vault_alerts(snapshot: V2GovernanceSnapshot, alerts: List[_VaultAlert]) -> None:
+    """Send the buffered sections as one Telegram message, or "(i/N)" parts if long.
+
+    No-op when ``alerts`` is empty so the caller doesn't have to guard. Every
+    part carries the same header and the highest severity of the whole group, so
+    a LOW section bundled with an owner change still pings the channel.
+    """
+    if not alerts:
+        return
+    severity = max((a.severity for a in alerts), key=_SEVERITY_ORDER.index)
+    header = _vault_header(snapshot)
+    parts = _split_into_messages(alerts, _MAX_MESSAGE_CHARS - len(header))
+    total = len(parts)
+    for index, bodies in enumerate(parts, start=1):
+        suffix = f" ({index}/{total})" if total > 1 else ""
+        message = f"{header}{suffix}\n\n" + _SECTION_SEPARATOR.join(bodies)
+        send_alert(Alert(severity, message, PROTOCOL))
+
+
+def _alert_pending_new(
+    snapshot: V2GovernanceSnapshot,
+    pc: PendingConfig,
+    operation_label: str,
+    diff: _VaultDiff,
+) -> None:
+    """Buffer a new-pending alert body; the caller flushes as one grouped message."""
+    diff.alert(
+        AlertSeverity.MEDIUM,
+        f"📥 Submitted: {operation_label}\n"
+        f"⏰ Executable at: {_format_ts(pc.valid_at)} {_format_countdown(pc.valid_at)}\n"
+        f"🔗 Tx: {_explorer_link(snapshot.chain, pc.tx_hash)}",
     )
 
 
 def _alert_pending_resolved(
-    snapshot: V2GovernanceSnapshot,
     data_hash: str,
     last_valid_at: int,
     function_name: str,
+    diff: _VaultDiff,
 ) -> None:
-    """Alert that a previously-pending operation no longer appears in pendingConfigs.
+    """Buffer a resolved-pending alert body.
 
     We can't always distinguish ``Accept`` from ``Revoke`` from a snapshot diff,
     but ``validAt`` gives a strong hint: if it has elapsed, the operation was
@@ -317,73 +412,46 @@ def _alert_pending_resolved(
     verb = "executed" if last_valid_at <= now else "revoked"
     icon = "✅" if verb == "executed" else "🛑"
     operation = f"`{function_name}()`" if function_name else f"`{data_hash[:10]}…`"
-    send_alert(
-        Alert(
-            AlertSeverity.LOW,
-            f"{icon} V2 [{snapshot.name}]({get_vault_url(snapshot.address, snapshot.chain)}) "
-            f"on {snapshot.chain.name}\n"
-            f"Pending operation {operation} was {verb} "
-            f"(was due {_format_ts(last_valid_at)}).",
-            PROTOCOL,
-        )
+    diff.alert(
+        AlertSeverity.LOW, f"{icon} Pending operation {operation} was {verb} (was due {_format_ts(last_valid_at)})."
     )
 
 
-def _alert_role_change(snapshot: V2GovernanceSnapshot, role: str, before: str, after: str) -> None:
+def _alert_role_change(role: str, before: str, after: str, diff: _VaultDiff) -> None:
     icon = "👑" if role == "owner" else "🎩"
-    send_alert(
-        Alert(
-            AlertSeverity.HIGH,
-            f"🚨 V2 [{snapshot.name}]({get_vault_url(snapshot.address, snapshot.chain)}) "
-            f"on {snapshot.chain.name}\n"
-            f"{icon} {role.capitalize()} changed: `{before}` → `{after}`",
-            PROTOCOL,
-        )
-    )
+    diff.alert(AlertSeverity.HIGH, f"🚨 {icon} {role.capitalize()} changed: `{before}` → `{after}`")
 
 
-def _alert_set_diff(
-    snapshot: V2GovernanceSnapshot,
-    set_name: str,
-    added: set[str],
-    removed: set[str],
-) -> None:
+def _alert_set_diff(set_name: str, added: set[str], removed: set[str], diff: _VaultDiff) -> None:
     icon = {"sentinels": "🛡️", "allocators": "🎯", "adapters": "🧩"}.get(set_name, "ℹ️")
     lines: list[str] = []
     for addr in sorted(added):
         lines.append(f"  + `{addr}`")
     for addr in sorted(removed):
         lines.append(f"  − `{addr}`")
-    send_alert(
-        Alert(
-            AlertSeverity.LOW,
-            f"{icon} V2 [{snapshot.name}]({get_vault_url(snapshot.address, snapshot.chain)}) "
-            f"{set_name} changed on {snapshot.chain.name}\n" + "\n".join(lines),
-            PROTOCOL,
-        )
-    )
+    diff.alert(AlertSeverity.LOW, f"{icon} {set_name} changed\n" + "\n".join(lines))
 
 
 # ----------------------------------------------------------------------------
 # Diff logic
 # ----------------------------------------------------------------------------
 
 
-def _diff_pending(snapshot: V2GovernanceSnapshot) -> None:
+def _diff_pending(snapshot: V2GovernanceSnapshot, diff: _VaultDiff) -> None:
     addr = snapshot.address.lower()
 
     current_keys: set[str] = set()
     for pc in snapshot.pending_configs:
         current_keys.add(pc.data_hash)
         operation_label = _operation_label(snapshot, pc)
-        _write(_pending_function_key(snapshot, pc.data_hash), _operation_function_name(pc, operation_label))
+        diff.write(_pending_function_key(snapshot, pc.data_hash), _operation_function_name(pc, operation_label))
         cache_key = morpho_key(addr, pc.data_hash, PENDING_TYPE)
         last = _read_int(cache_key)
         # Already alerted at this validAt, or marked executed.
         if last == pc.valid_at or last == EXECUTED:
             continue
-        _alert_pending_new(snapshot, pc, operation_label)
-        _write(cache_key, pc.valid_at)
+        _alert_pending_new(snapshot, pc, operation_label, diff)
+        diff.write(cache_key, pc.valid_at)
 
     # Detect resolved entries: anything in last-run's index that isn't in the
     # current pending list.
@@ -397,22 +465,27 @@ def _diff_pending(snapshot: V2GovernanceSnapshot) -> None:
         if last <= 0:
             # Already marked executed/revoked.
             continue
-        _alert_pending_resolved(snapshot, data_hash, last, _read_str(_pending_function_key(snapshot, data_hash)))
-        _write(cache_key, EXECUTED if last <= int(datetime.now().timestamp()) else REVOKED)
+        _alert_pending_resolved(
+            data_hash,
+            last,
+            _read_str(_pending_function_key(snapshot, data_hash)),
+            diff,
+        )
+        diff.write(cache_key, EXECUTED if last <= int(datetime.now().timestamp()) else REVOKED)
 
-    _write(index_key, ",".join(sorted(current_keys)))
+    diff.write(index_key, ",".join(sorted(current_keys)))
 
 
-def _diff_single_role(snapshot: V2GovernanceSnapshot, role: str, current: str) -> None:
+def _diff_single_role(snapshot: V2GovernanceSnapshot, role: str, current: str, diff: _VaultDiff) -> None:
     cache_key = morpho_key(snapshot.address.lower(), role, ROLE_TYPE)
     last = _read_str(cache_key)
     cur_lc = current.lower()
     if last and last != cur_lc:
-        _alert_role_change(snapshot, role, last, current)
-    _write(cache_key, cur_lc)
+        _alert_role_change(role, last, current, diff)
+    diff.write(cache_key, cur_lc)
 
 
-def _diff_set(snapshot: V2GovernanceSnapshot, set_name: str, current: List[str]) -> None:
+def _diff_set(snapshot: V2GovernanceSnapshot, set_name: str, current: List[str], diff: _VaultDiff) -> None:
     cache_key = morpho_key(snapshot.address.lower(), set_name, SET_TYPE)
     last_str = _read_str(cache_key)
     last_set = {a for a in last_str.split(",") if a} if last_str else set()
@@ -423,18 +496,35 @@ def _diff_set(snapshot: V2GovernanceSnapshot, set_name: str, current: List[str])
     if last_str and (added or removed):
         added_cs: set[str] = {str(Web3.to_checksum_address(a)) for a in added}
         removed_cs: set[str] = {str(Web3.to_checksum_address(a)) for a in removed}
-        _alert_set_diff(snapshot, set_name, added_cs, removed_cs)
-    _write(cache_key, ",".join(sorted(current_set)))
+        _alert_set_diff(set_name, added_cs, removed_cs, diff)
+    diff.write(cache_key, ",".join(sorted(current_set)))
 
 
 def diff_and_alert(snapshot: V2GovernanceSnapshot) -> None:
-    """Diff a vault's snapshot against persisted state and emit Telegram alerts."""
-    _diff_pending(snapshot)
-    _diff_single_role(snapshot, "owner", snapshot.owner)
-    _diff_single_role(snapshot, "curator", snapshot.curator)
-    _diff_set(snapshot, "sentinels", snapshot.sentinels)
-    _diff_set(snapshot, "allocators", snapshot.allocators)
-    _diff_set(snapshot, "adapters", snapshot.adapters)
+    """Diff a vault's snapshot against persisted state and emit one grouped alert.
+
+    Every diff category (pending, owner/curator, sentinels/allocators/adapters)
+    appends to a per-vault buffer. We then send a single Telegram message with
+    one header (``V2 [name](url) on chain``) and the highest severity of the
+    group, so a vault with several simultaneous changes doesn't spam N separate
+    messages; only a group too long for one Telegram message is split into
+    numbered parts.
+
+    Cache cursors are committed after the send, not during the diff: if Telegram
+    is down, the next run re-detects the same changes and re-alerts rather than
+    treating them as already delivered. A partial send (part 1 of 3 lands, part 2
+    fails) therefore repeats the whole group next run — duplicates beat a
+    governance change nobody ever sees.
+    """
+    diff = _VaultDiff()
+    _diff_pending(snapshot, diff)
+    _diff_single_role(snapshot, "owner", snapshot.owner, diff)
+    _diff_single_role(snapshot, "curator", snapshot.curator, diff)
+    _diff_set(snapshot, "sentinels", snapshot.sentinels, diff)
+    _diff_set(snapshot, "allocators", snapshot.allocators, diff)
+    _diff_set(snapshot, "adapters", snapshot.adapters, diff)
+    _send_vault_alerts(snapshot, diff.alerts)
+    diff.commit()
 
 
 # ----------------------------------------------------------------------------