From 4e764e3ef8a12fd9e606b0557cdeefb31b3d84cd Mon Sep 17 00:00:00 2001 From: yvtapir Date: Fri, 28 Aug 2026 16:17:40 +0300 Subject: [PATCH 1/3] feat(yearn): monitor small parent vault deposits --- automation/jobs.yaml | 1 + monitoring.yaml | 3 + protocols/yearn/README.md | 25 ++ .../yearn/alert_small_parent_deposits.py | 351 ++++++++++++++++++ protocols/yearn/kong.py | 79 ++++ tests/test_small_parent_deposits.py | 116 ++++++ tests/test_yearn_kong.py | 59 +++ 7 files changed, 634 insertions(+) create mode 100644 protocols/yearn/alert_small_parent_deposits.py create mode 100644 tests/test_small_parent_deposits.py diff --git a/automation/jobs.yaml b/automation/jobs.yaml index b60f0f54..fbeb6ee4 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -61,6 +61,7 @@ profiles: - { name: "stables-oracles", script: protocols/stables/oracles.py } - { name: "cap-status", script: protocols/cap/status.py } - { name: "yearn-alert-large-flows", script: protocols/yearn/alert_large_flows.py } + - { name: "yearn-alert-small-parent-deposits", script: protocols/yearn/alert_small_parent_deposits.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 } - { name: "maple", script: protocols/maple/main.py, enabled: false } diff --git a/monitoring.yaml b/monitoring.yaml index 09a67552..a5d015aa 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -460,6 +460,7 @@ protocols: tasks: - protocols/yearn/lender_borrower.py - protocols/yearn/alert_large_flows.py + - protocols/yearn/alert_small_parent_deposits.py - protocols/yearn/check_timelock_delay.py - protocols/yearn/check_indexer_freshness.py monitors: @@ -470,6 +471,8 @@ protocols: severity: "MEDIUM" - name: "Large Flows" description: "Deposit/withdrawal flows >=$500k USD (Katana withdrawals >=$50k; or 10% of vault totalSupply fallback for unpriced tokens)" + - name: "Small Parent Vault Deposits" + description: "Individual deposits below 10,000 normalized underlying-token units into active Yearn v3 parent vaults" - name: "Timelock Delay" description: "Yearn TimelockController getMinDelay() < 7 days across Mainnet, Base, Arbitrum, Polygon, Optimism, Katana" - name: "Timelock Events" diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 27a5d092..4a93dc25 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -55,6 +55,31 @@ Optional flags: - `--chain-ids` (default: all vault chain IDs — `1,8453,42161,747474`) - `--no-cache` (disable caching) +## Small Parent Vault Deposits + +The script `yearn/alert_small_parent_deposits.py` alerts on every positive deposit strictly below 10,000 normalized underlying-token units into an active Yearn v3 parent vault. The comparison is in token units, not USD: for example, both 9,999 USDC and 9,999 WETH qualify. + +### Data Sources + +- **Parent vault discovery**: Kong GraphQL, filtered to Yearn v3 `vaultType: 1` vaults and excluding retired or hidden entries. +- **Deposit events**: Envio `Deposit` entities, including the ERC-4626 owner and sender as well as the transaction initiator. +- **Token decimals**: the parent vault's underlying asset metadata from Kong. + +Events are processed with a per-chain `(blockNumber, logIndex)` cursor stored in the monitoring database. The cursor advances only after an event is successfully evaluated and, when applicable, delivered to Telegram. A new deployment starts with a two-hour lookback. + +### Usage + +```bash +uv run protocols/yearn/alert_small_parent_deposits.py +``` + +Optional flags: + +- `--threshold-units` (default: `10000`) +- `--lookback-seconds` (default: `7200`, used only before a chain cursor exists) +- `--page-size` (default: `1000`) +- `--chain-ids` (default: `1,10,8453,42161,137,747474`) + ======= ## Shadow Debt Check diff --git a/protocols/yearn/alert_small_parent_deposits.py b/protocols/yearn/alert_small_parent_deposits.py new file mode 100644 index 00000000..f8ec64d3 --- /dev/null +++ b/protocols/yearn/alert_small_parent_deposits.py @@ -0,0 +1,351 @@ +#!/usr/bin/env python3 +"""Alert on deposits below a token-unit threshold into Yearn v3 parent vaults.""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +import sys +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from decimal import Decimal, getcontext +from typing import Callable + +from dotenv import load_dotenv + +from protocols.yearn.kong import fetch_kong_parent_vaults +from utils import store +from utils.alert import Alert, AlertSeverity, send_alert +from utils.chains import EXPLORER_URLS, Chain +from utils.logger import get_logger +from utils.telegram import send_envio_error_message + +load_dotenv() + +getcontext().prec = 60 + +ENVIO_GRAPHQL_URL = os.getenv("ENVIO_GRAPHQL_URL") +DEFAULT_LOG_LEVEL = os.getenv("SMALL_PARENT_DEPOSITS_LOG_LEVEL", "INFO") +DEFAULT_THRESHOLD_UNITS = Decimal("10000") +DEFAULT_LOOKBACK_SECONDS = 7200 +DEFAULT_PAGE_SIZE = 1000 +PROTOCOL = "yearn" +STATE_NAMESPACE = "yearn.small_parent_deposits" + +logger = get_logger("yearn.alert_small_parent_deposits") + + +@dataclass(frozen=True, order=True) +class EventCursor: + """Per-chain Envio event cursor.""" + + block_number: int + log_index: int + + +def http_json(url: str, body: dict) -> dict: + """POST a JSON body and return the decoded response.""" + request = urllib.request.Request( + url, + data=json.dumps(body).encode("utf-8"), + headers={"Accept": "application/json", "Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=30) as response: + payload: object = json.loads(response.read().decode("utf-8")) + if not isinstance(payload, dict): + raise ValueError("Envio returned a non-object JSON response") + return payload + + +def gql_request(query: str, variables: dict) -> dict | None: + """Execute an Envio GraphQL query, routing failures to its ops channel.""" + if not ENVIO_GRAPHQL_URL: + raise RuntimeError("ENVIO_GRAPHQL_URL is not set") + + try: + payload = http_json(ENVIO_GRAPHQL_URL, {"query": query, "variables": variables}) + except (urllib.error.HTTPError, urllib.error.URLError, ConnectionError, OSError, ValueError) as exc: + send_envio_error_message( + f"Small parent deposit monitor: Envio GraphQL request failed ({exc}). Skipping this run.", + PROTOCOL, + source="small_parent_deposits", + ) + logger.error("Envio request failed: %s", exc) + return None + + if payload.get("errors"): + send_envio_error_message( + f"Small parent deposit monitor: Envio GraphQL errors: {payload['errors']}", + PROTOCOL, + source="small_parent_deposits", + ) + logger.error("Envio GraphQL errors: %s", payload["errors"]) + return None + return payload + + +def load_deposits( + chain_id: int, + vault_addresses: list[str], + cursor: EventCursor, + since_ts: int, + limit: int, +) -> list[dict] | None: + """Load one ordered page of parent-vault deposits after ``cursor``.""" + query = """ + query SmallParentDeposits( + $chainId: Int! + $addresses: [String!]! + $lastBlock: Int! + $lastLogIndex: Int! + $sinceTs: Int! + $limit: Int! + ) { + Deposit( + where: { + chainId: { _eq: $chainId } + vaultAddress: { _in: $addresses } + _or: [ + { blockNumber: { _gt: $lastBlock }, blockTimestamp: { _gte: $sinceTs } } + { blockNumber: { _eq: $lastBlock }, logIndex: { _gt: $lastLogIndex } } + ] + } + order_by: { blockNumber: asc, logIndex: asc } + limit: $limit + ) { + id + vaultAddress + chainId + blockNumber + blockTimestamp + transactionHash + transactionFrom + logIndex + sender + owner + assets + shares + } + } + """ + variables = { + "chainId": chain_id, + "addresses": vault_addresses, + "lastBlock": cursor.block_number, + "lastLogIndex": cursor.log_index, + "sinceTs": since_ts, + "limit": limit, + } + response = gql_request(query, variables) + if response is None: + return None + deposits = response.get("data", {}).get("Deposit") + if not isinstance(deposits, list): + raise RuntimeError("Envio response missing Deposit list") + return deposits + + +def format_units(raw_assets: str | int, decimals: int) -> Decimal: + """Convert an integer asset amount into normalized token units.""" + return Decimal(str(raw_assets)) / (Decimal(10) ** decimals) + + +def is_small_deposit(raw_assets: str | int, decimals: int, threshold_units: Decimal) -> bool: + """Return whether a positive deposit is strictly below the unit threshold.""" + amount = format_units(raw_assets, decimals) + return Decimal(0) < amount < threshold_units + + +def format_amount(amount: Decimal) -> str: + """Format a token amount without scientific notation or trailing zeroes.""" + rendered = f"{amount:,.18f}".rstrip("0").rstrip(".") + return rendered or "0" + + +def address_link(address: str, explorer: str | None) -> str: + """Return a full address, linked to the chain explorer when available.""" + if explorer: + return f"[{address}]({explorer}/address/{address})" + return address + + +def build_alert_message(event: dict, vault: dict, amount: Decimal, threshold_units: Decimal) -> str: + """Build the Telegram message for one qualifying deposit.""" + chain_id = int(event["chainId"]) + chain = Chain.from_chain_id(chain_id) + explorer = EXPLORER_URLS.get(chain_id) + vault_address = str(event["vaultAddress"]) + tx_hash = str(event["transactionHash"]) + tx = f"[{tx_hash}]({explorer}/tx/{tx_hash})" if explorer else tx_hash + + lines = [ + "Small parent-vault deposit", + f"🏦 Vault: {address_link(vault_address, explorer)} ({vault['symbol']})", + f"🪙 Amount: {format_amount(amount)} {vault['asset_symbol']}", + f"📏 Threshold: < {format_amount(threshold_units)} {vault['asset_symbol']}", + f"⛓️ Chain: {chain.network_name}", + f"👤 Owner: {address_link(str(event['owner']), explorer)}", + f"💳 Sender: {address_link(str(event['sender']), explorer)}", + ] + transaction_from = event.get("transactionFrom") + if transaction_from: + lines.append(f"🚀 Tx From: {address_link(str(transaction_from), explorer)}") + lines.append(f"🔗 Tx: {tx}") + return "\n".join(lines) + + +def cursor_from_event(event: dict) -> EventCursor: + """Return the sortable cursor represented by an Envio event.""" + return EventCursor(int(event["blockNumber"]), int(event["logIndex"])) + + +def load_cursor(chain_id: int) -> EventCursor | None: + """Load a chain cursor from persistent monitor state.""" + raw = store.state_get(STATE_NAMESPACE, str(chain_id)) + if raw is None: + return None + try: + payload = json.loads(raw) + return EventCursor(int(payload["block_number"]), int(payload["log_index"])) + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise RuntimeError(f"Invalid small-deposit cursor for chain {chain_id}: {raw}") from exc + + +def save_cursor(chain_id: int, cursor: EventCursor) -> None: + """Persist a successfully processed chain cursor.""" + store.state_set( + STATE_NAMESPACE, + str(chain_id), + json.dumps({"block_number": cursor.block_number, "log_index": cursor.log_index}), + ) + + +def process_event( + event: dict, + vaults_by_address: dict[str, dict], + threshold_units: Decimal, + alert_sender: Callable[[Alert], None] = send_alert, +) -> bool: + """Evaluate one deposit, sending an alert when it is below the threshold.""" + vault_address = str(event["vaultAddress"]).lower() + vault = vaults_by_address.get(vault_address) + if vault is None: + raise RuntimeError(f"Envio returned unknown parent vault {event['vaultAddress']}") + + decimals = int(vault["asset_decimals"]) + if not is_small_deposit(event["assets"], decimals, threshold_units): + return False + + amount = format_units(event["assets"], decimals) + message = build_alert_message(event, vault, amount, threshold_units) + alert_sender(Alert(AlertSeverity.LOW, message, PROTOCOL)) + return True + + +def monitor_chain( + chain: Chain, + threshold_units: Decimal, + lookback_seconds: int, + page_size: int, + now: int | None = None, +) -> tuple[int, int]: + """Fetch and process all new deposits for one chain.""" + vaults = fetch_kong_parent_vaults(chain) + if not vaults: + logger.warning("No active parent vaults returned for %s", chain.network_name) + return 0, 0 + + vaults_by_address = {str(vault["address"]).lower(): vault for vault in vaults} + # Envio stores checksummed addresses, while older rows or deployments may + # use lowercase. Supplying both forms keeps the case-sensitive filter safe. + addresses = sorted( + {address for vault in vaults for address in (str(vault["address"]), str(vault["address"]).lower())} + ) + + persisted_cursor = load_cursor(chain.chain_id) + cursor = persisted_cursor or EventCursor(0, -1) + since_ts = 0 if persisted_cursor else (now or int(time.time())) - lookback_seconds + processed = 0 + alerted = 0 + + while True: + events = load_deposits(chain.chain_id, addresses, cursor, since_ts, page_size) + if events is None: + break + if not events: + break + + for event in events: + event_cursor = cursor_from_event(event) + if event_cursor <= cursor: + continue + if process_event(event, vaults_by_address, threshold_units): + alerted += 1 + save_cursor(chain.chain_id, event_cursor) + cursor = event_cursor + processed += 1 + + if len(events) < page_size: + break + + return processed, alerted + + +def parse_chain_ids(raw: str) -> list[Chain]: + """Parse a comma-separated chain-ID list.""" + chains: list[Chain] = [] + for value in raw.split(","): + if value.strip(): + chains.append(Chain.from_chain_id(int(value.strip()))) + return chains + + +def main() -> None: + """Run the small parent-vault deposit monitor.""" + default_chain_ids = ",".join(str(chain.chain_id) for chain in Chain) + parser = argparse.ArgumentParser( + description="Alert on Yearn v3 parent-vault deposits below a token-unit threshold." + ) + parser.add_argument("--threshold-units", type=Decimal, default=DEFAULT_THRESHOLD_UNITS) + parser.add_argument("--lookback-seconds", type=int, default=DEFAULT_LOOKBACK_SECONDS) + parser.add_argument("--page-size", type=int, default=DEFAULT_PAGE_SIZE) + parser.add_argument("--chain-ids", default=default_chain_ids) + parser.add_argument("--log-level", default=DEFAULT_LOG_LEVEL) + args = parser.parse_args() + + logging.basicConfig( + level=args.log_level.upper(), + format="[%(name)s] %(levelname)s %(message)s", + stream=sys.stderr, + ) + if args.threshold_units <= 0: + parser.error("--threshold-units must be positive") + if args.lookback_seconds < 0: + parser.error("--lookback-seconds must be non-negative") + if args.page_size <= 0: + parser.error("--page-size must be positive") + + total_processed = 0 + total_alerted = 0 + for chain in parse_chain_ids(args.chain_ids): + processed, alerted = monitor_chain( + chain, + args.threshold_units, + args.lookback_seconds, + args.page_size, + ) + total_processed += processed + total_alerted += alerted + logger.info("%s: processed=%d alerted=%d", chain.network_name, processed, alerted) + logger.info("complete: processed=%d alerted=%d", total_processed, total_alerted) + + +if __name__ == "__main__": + from utils.runner import run_with_alert + + run_with_alert(main, PROTOCOL) diff --git a/protocols/yearn/kong.py b/protocols/yearn/kong.py index 279d016d..2f20b948 100644 --- a/protocols/yearn/kong.py +++ b/protocols/yearn/kong.py @@ -21,6 +21,26 @@ } } """ +KONG_PARENT_VAULTS_QUERY = """ +query YearnParentVaults($chainId: Int) { + vaults(chainId: $chainId, v3: true, yearn: true, vaultType: 1) { + address + name + symbol + decimals + vaultType + asset { + address + symbol + decimals + } + meta { + isRetired + isHidden + } + } +} +""" STRATEGY_SOURCE_ALL = "strategies" STRATEGY_SOURCE_DEFAULT_QUEUE = "default_queue" @@ -86,6 +106,14 @@ def _is_retired(vault: Dict[str, Any]) -> bool: return bool(meta.get("isRetired")) +def _is_hidden(vault: Dict[str, Any]) -> bool: + """Return whether Kong metadata marks the vault hidden.""" + meta = vault.get("meta") + if not isinstance(meta, dict): + return False + return bool(meta.get("isHidden")) + + def _strategy_field(strategy_source: str) -> str: """Return the Kong field backing the requested strategy source.""" if strategy_source == STRATEGY_SOURCE_ALL: @@ -138,3 +166,54 @@ def fetch_kong_vaults( ) return result + + +def fetch_kong_parent_vaults(chain: Chain) -> List[Dict[str, object]]: + """Fetch active Yearn v3 parent/allocator vault metadata from Kong. + + Kong's ``vaultType: 1`` identifies parent/allocator vaults, while + ``vaultType: 2`` identifies strategy vaults. Retired and hidden vaults are + excluded locally so callers only monitor active user-facing parents. + + Args: + chain: Chain to fetch. + + Returns: + Parent vault dicts containing vault and underlying-asset metadata. + + Raises: + KongRequestError: If Kong omits required address or decimal metadata. + """ + data = _post_graphql(KONG_PARENT_VAULTS_QUERY, {"chainId": chain.chain_id}) + vaults = data.get("vaults") + if not isinstance(vaults, list): + raise KongRequestError("Kong response missing vaults list") + + result: List[Dict[str, object]] = [] + for vault in vaults: + if not isinstance(vault, dict) or _is_retired(vault) or _is_hidden(vault): + continue + + address = vault.get("address") + asset = vault.get("asset") + if not isinstance(address, str) or not isinstance(asset, dict): + raise KongRequestError("Kong parent vault missing address or asset metadata") + + asset_address = asset.get("address") + asset_decimals = _parse_decimals(asset.get("decimals")) + if not isinstance(asset_address, str) or asset_decimals is None: + raise KongRequestError(f"Kong parent vault {address} missing asset address or decimals") + + result.append( + { + "address": address, + "name": vault.get("name") or vault.get("symbol") or "UNKNOWN", + "symbol": vault.get("symbol") or "UNKNOWN", + "decimals": _parse_decimals(vault.get("decimals")), + "asset_address": asset_address, + "asset_symbol": asset.get("symbol") or "UNKNOWN", + "asset_decimals": asset_decimals, + } + ) + + return result diff --git a/tests/test_small_parent_deposits.py b/tests/test_small_parent_deposits.py new file mode 100644 index 00000000..f69ce7ee --- /dev/null +++ b/tests/test_small_parent_deposits.py @@ -0,0 +1,116 @@ +from decimal import Decimal + +from protocols.yearn import alert_small_parent_deposits as monitor +from utils.alert import AlertSeverity +from utils.chains import Chain + +VAULT = { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": 6, + "asset_address": "0xAsset", + "asset_symbol": "USDC", + "asset_decimals": 6, +} + + +def make_event(*, assets: str = "9999999999", block_number: int = 100, log_index: int = 2) -> dict: + return { + "id": f"1_{block_number}_{log_index}", + "vaultAddress": "0xParent", + "chainId": 1, + "blockNumber": block_number, + "blockTimestamp": 1_700_000_000, + "transactionHash": "0xTransaction", + "transactionFrom": "0xTransactionFrom", + "logIndex": log_index, + "sender": "0xSender", + "owner": "0xOwner", + "assets": assets, + "shares": assets, + } + + +def test_small_deposit_uses_normalized_token_units() -> None: + threshold = Decimal("10000") + + assert monitor.format_units("1234567", 6) == Decimal("1.234567") + assert monitor.is_small_deposit("9999999999", 6, threshold) + assert not monitor.is_small_deposit("10000000000", 6, threshold) + assert not monitor.is_small_deposit("10000000001", 6, threshold) + assert not monitor.is_small_deposit("0", 6, threshold) + + +def test_process_event_sends_low_alert_with_all_addresses() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(), + {"0xparent": VAULT}, + Decimal("10000"), + alert_sender=alerts.append, + ) + + assert did_alert + assert len(alerts) == 1 + alert = alerts[0] + assert alert.severity is AlertSeverity.LOW + assert alert.protocol == "yearn" + assert "9,999.999999 USDC" in alert.message + assert "0xOwner" in alert.message + assert "0xSender" in alert.message + assert "0xTransactionFrom" in alert.message + assert "0xTransaction" in alert.message + + +def test_process_event_does_not_alert_at_threshold() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(assets="10000000000"), + {"0xparent": VAULT}, + Decimal("10000"), + alert_sender=alerts.append, + ) + + assert not did_alert + assert alerts == [] + + +def test_monitor_chain_pages_and_persists_each_processed_event(monkeypatch) -> None: + first = make_event(block_number=100, log_index=2) + second = make_event(block_number=101, log_index=3) + calls = [] + saved = [] + + monkeypatch.setattr(monitor, "fetch_kong_parent_vaults", lambda _chain: [VAULT]) + monkeypatch.setattr(monitor, "load_cursor", lambda _chain_id: None) + monkeypatch.setattr(monitor, "save_cursor", lambda chain_id, cursor: saved.append((chain_id, cursor))) + monkeypatch.setattr(monitor, "process_event", lambda event, *_args: event is first) + + def fake_load(chain_id, addresses, cursor, since_ts, limit): + calls.append((chain_id, addresses, cursor, since_ts, limit)) + if len(calls) == 1: + return [first, second] + return [] + + monkeypatch.setattr(monitor, "load_deposits", fake_load) + + processed, alerted = monitor.monitor_chain( + Chain.MAINNET, + Decimal("10000"), + lookback_seconds=7200, + page_size=2, + now=1_700_010_000, + ) + + assert (processed, alerted) == (2, 1) + assert calls[0][0] == 1 + assert calls[0][2] == monitor.EventCursor(0, -1) + assert calls[0][3] == 1_700_002_800 + assert calls[1][2] == monitor.EventCursor(101, 3) + assert saved == [ + (1, monitor.EventCursor(100, 2)), + (1, monitor.EventCursor(101, 3)), + ] diff --git a/tests/test_yearn_kong.py b/tests/test_yearn_kong.py index 4a64d615..a7442134 100644 --- a/tests/test_yearn_kong.py +++ b/tests/test_yearn_kong.py @@ -87,3 +87,62 @@ def test_fetch_kong_vaults_raises_on_graphql_errors(monkeypatch) -> None: with pytest.raises(kong.KongRequestError): kong.fetch_kong_vaults(Chain.MAINNET) + + +def test_fetch_kong_parent_vaults_filters_inactive_vaults(monkeypatch) -> None: + payload = { + "data": { + "vaults": [ + { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": "6", + "vaultType": 1, + "asset": {"address": "0xAsset", "symbol": "USDC", "decimals": 6}, + "meta": {"isRetired": False, "isHidden": False}, + }, + { + "address": "0xRetired", + "name": "Old vault", + "symbol": "yvOLD", + "decimals": "18", + "vaultType": 1, + "asset": {"address": "0xOldAsset", "symbol": "OLD", "decimals": 18}, + "meta": {"isRetired": True, "isHidden": False}, + }, + { + "address": "0xHidden", + "name": "Hidden vault", + "symbol": "yvHIDDEN", + "decimals": "18", + "vaultType": 1, + "asset": {"address": "0xHiddenAsset", "symbol": "HIDDEN", "decimals": 18}, + "meta": {"isRetired": False, "isHidden": True}, + }, + ] + } + } + calls = [] + + def fake_post(url: str, json: dict, timeout: int) -> FakeResponse: + calls.append((url, json, timeout)) + return FakeResponse(payload) + + monkeypatch.setattr(kong.requests, "post", fake_post) + + vaults = kong.fetch_kong_parent_vaults(Chain.MAINNET) + + assert "vaultType: 1" in calls[0][1]["query"] + assert calls[0][1]["variables"] == {"chainId": 1} + assert vaults == [ + { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": 6, + "asset_address": "0xAsset", + "asset_symbol": "USDC", + "asset_decimals": 6, + } + ] From e2406d267454fcde2467fa1be436141c98cc7a82 Mon Sep 17 00:00:00 2001 From: yvtapir Date: Fri, 28 Aug 2026 16:24:14 +0300 Subject: [PATCH 2/3] feat(yearn): monitor small parent vault withdrawals --- automation/jobs.yaml | 2 +- monitoring.yaml | 6 +- protocols/yearn/README.md | 10 +- ...eposits.py => alert_small_parent_flows.py} | 157 +++++++++----- tests/test_small_parent_deposits.py | 116 ----------- tests/test_small_parent_flows.py | 193 ++++++++++++++++++ 6 files changed, 310 insertions(+), 174 deletions(-) rename protocols/yearn/{alert_small_parent_deposits.py => alert_small_parent_flows.py} (70%) delete mode 100644 tests/test_small_parent_deposits.py create mode 100644 tests/test_small_parent_flows.py diff --git a/automation/jobs.yaml b/automation/jobs.yaml index fbeb6ee4..9db0ea9c 100644 --- a/automation/jobs.yaml +++ b/automation/jobs.yaml @@ -61,7 +61,7 @@ profiles: - { name: "stables-oracles", script: protocols/stables/oracles.py } - { name: "cap-status", script: protocols/cap/status.py } - { name: "yearn-alert-large-flows", script: protocols/yearn/alert_large_flows.py } - - { name: "yearn-alert-small-parent-deposits", script: protocols/yearn/alert_small_parent_deposits.py } + - { name: "yearn-alert-small-parent-flows", script: protocols/yearn/alert_small_parent_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 } - { name: "maple", script: protocols/maple/main.py, enabled: false } diff --git a/monitoring.yaml b/monitoring.yaml index a5d015aa..578d8b7f 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -460,7 +460,7 @@ protocols: tasks: - protocols/yearn/lender_borrower.py - protocols/yearn/alert_large_flows.py - - protocols/yearn/alert_small_parent_deposits.py + - protocols/yearn/alert_small_parent_flows.py - protocols/yearn/check_timelock_delay.py - protocols/yearn/check_indexer_freshness.py monitors: @@ -471,8 +471,8 @@ protocols: severity: "MEDIUM" - name: "Large Flows" description: "Deposit/withdrawal flows >=$500k USD (Katana withdrawals >=$50k; or 10% of vault totalSupply fallback for unpriced tokens)" - - name: "Small Parent Vault Deposits" - description: "Individual deposits below 10,000 normalized underlying-token units into active Yearn v3 parent vaults" + - name: "Small Parent Vault Flows" + description: "Individual deposits and withdrawals below 10,000 normalized underlying-token units for active Yearn v3 parent vaults" - name: "Timelock Delay" description: "Yearn TimelockController getMinDelay() < 7 days across Mainnet, Base, Arbitrum, Polygon, Optimism, Katana" - name: "Timelock Events" diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 4a93dc25..15278372 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -55,22 +55,22 @@ Optional flags: - `--chain-ids` (default: all vault chain IDs — `1,8453,42161,747474`) - `--no-cache` (disable caching) -## Small Parent Vault Deposits +## Small Parent Vault Flows -The script `yearn/alert_small_parent_deposits.py` alerts on every positive deposit strictly below 10,000 normalized underlying-token units into an active Yearn v3 parent vault. The comparison is in token units, not USD: for example, both 9,999 USDC and 9,999 WETH qualify. +The script `yearn/alert_small_parent_flows.py` alerts on every positive deposit or withdrawal strictly below 10,000 normalized underlying-token units for an active Yearn v3 parent vault. The comparison is in token units, not USD: for example, both 9,999 USDC and 9,999 WETH qualify. ### Data Sources - **Parent vault discovery**: Kong GraphQL, filtered to Yearn v3 `vaultType: 1` vaults and excluding retired or hidden entries. -- **Deposit events**: Envio `Deposit` entities, including the ERC-4626 owner and sender as well as the transaction initiator. +- **Flow events**: Envio `Deposit` and `Withdraw` entities. Alerts include the ERC-4626 owner and sender, the transaction initiator, and the asset receiver for withdrawals. - **Token decimals**: the parent vault's underlying asset metadata from Kong. -Events are processed with a per-chain `(blockNumber, logIndex)` cursor stored in the monitoring database. The cursor advances only after an event is successfully evaluated and, when applicable, delivered to Telegram. A new deployment starts with a two-hour lookback. +Deposits and withdrawals are processed with independent per-chain `(blockNumber, logIndex)` cursors stored in the monitoring database. A cursor advances only after an event is successfully evaluated and, when applicable, delivered to Telegram. A new deployment starts each stream with a two-hour lookback. ### Usage ```bash -uv run protocols/yearn/alert_small_parent_deposits.py +uv run protocols/yearn/alert_small_parent_flows.py ``` Optional flags: diff --git a/protocols/yearn/alert_small_parent_deposits.py b/protocols/yearn/alert_small_parent_flows.py similarity index 70% rename from protocols/yearn/alert_small_parent_deposits.py rename to protocols/yearn/alert_small_parent_flows.py index f8ec64d3..4a685cb5 100644 --- a/protocols/yearn/alert_small_parent_deposits.py +++ b/protocols/yearn/alert_small_parent_flows.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Alert on deposits below a token-unit threshold into Yearn v3 parent vaults.""" +"""Alert on small deposits and withdrawals from Yearn v3 parent vaults.""" from __future__ import annotations @@ -29,14 +29,16 @@ getcontext().prec = 60 ENVIO_GRAPHQL_URL = os.getenv("ENVIO_GRAPHQL_URL") -DEFAULT_LOG_LEVEL = os.getenv("SMALL_PARENT_DEPOSITS_LOG_LEVEL", "INFO") +DEFAULT_LOG_LEVEL = os.getenv("SMALL_PARENT_FLOWS_LOG_LEVEL", "INFO") DEFAULT_THRESHOLD_UNITS = Decimal("10000") DEFAULT_LOOKBACK_SECONDS = 7200 DEFAULT_PAGE_SIZE = 1000 PROTOCOL = "yearn" -STATE_NAMESPACE = "yearn.small_parent_deposits" +STATE_NAMESPACE = "yearn.small_parent_flows" +FLOW_TYPES = ("deposit", "withdrawal") +FLOW_ENTITY = {"deposit": "Deposit", "withdrawal": "Withdraw"} -logger = get_logger("yearn.alert_small_parent_deposits") +logger = get_logger("yearn.alert_small_parent_flows") @dataclass(frozen=True, order=True) @@ -71,34 +73,41 @@ def gql_request(query: str, variables: dict) -> dict | None: payload = http_json(ENVIO_GRAPHQL_URL, {"query": query, "variables": variables}) except (urllib.error.HTTPError, urllib.error.URLError, ConnectionError, OSError, ValueError) as exc: send_envio_error_message( - f"Small parent deposit monitor: Envio GraphQL request failed ({exc}). Skipping this run.", + f"Small parent flow monitor: Envio GraphQL request failed ({exc}). Skipping this run.", PROTOCOL, - source="small_parent_deposits", + source="small_parent_flows", ) logger.error("Envio request failed: %s", exc) return None if payload.get("errors"): send_envio_error_message( - f"Small parent deposit monitor: Envio GraphQL errors: {payload['errors']}", + f"Small parent flow monitor: Envio GraphQL errors: {payload['errors']}", PROTOCOL, - source="small_parent_deposits", + source="small_parent_flows", ) logger.error("Envio GraphQL errors: %s", payload["errors"]) return None return payload -def load_deposits( +def load_events( + flow_type: str, chain_id: int, vault_addresses: list[str], cursor: EventCursor, since_ts: int, limit: int, ) -> list[dict] | None: - """Load one ordered page of parent-vault deposits after ``cursor``.""" + """Load one ordered page of parent-vault flow events after ``cursor``.""" + try: + entity = FLOW_ENTITY[flow_type] + except KeyError as exc: + raise ValueError(f"Unknown flow type: {flow_type}") from exc + + receiver_field = "receiver" if flow_type == "withdrawal" else "" query = """ - query SmallParentDeposits( + query SmallParentFlows( $chainId: Int! $addresses: [String!]! $lastBlock: Int! @@ -106,7 +115,7 @@ def load_deposits( $sinceTs: Int! $limit: Int! ) { - Deposit( + events: __ENTITY__( where: { chainId: { _eq: $chainId } vaultAddress: { _in: $addresses } @@ -128,11 +137,13 @@ def load_deposits( logIndex sender owner + __RECEIVER_FIELD__ assets shares } } """ + query = query.replace("__ENTITY__", entity).replace("__RECEIVER_FIELD__", receiver_field) variables = { "chainId": chain_id, "addresses": vault_addresses, @@ -144,10 +155,10 @@ def load_deposits( response = gql_request(query, variables) if response is None: return None - deposits = response.get("data", {}).get("Deposit") - if not isinstance(deposits, list): - raise RuntimeError("Envio response missing Deposit list") - return deposits + events = response.get("data", {}).get("events") + if not isinstance(events, list): + raise RuntimeError(f"Envio response missing {entity} list") + return [{**event, "flow_type": flow_type} for event in events] def format_units(raw_assets: str | int, decimals: int) -> Decimal: @@ -155,8 +166,8 @@ def format_units(raw_assets: str | int, decimals: int) -> Decimal: return Decimal(str(raw_assets)) / (Decimal(10) ** decimals) -def is_small_deposit(raw_assets: str | int, decimals: int, threshold_units: Decimal) -> bool: - """Return whether a positive deposit is strictly below the unit threshold.""" +def is_small_flow(raw_assets: str | int, decimals: int, threshold_units: Decimal) -> bool: + """Return whether a positive flow is strictly below the unit threshold.""" amount = format_units(raw_assets, decimals) return Decimal(0) < amount < threshold_units @@ -175,16 +186,17 @@ def address_link(address: str, explorer: str | None) -> str: def build_alert_message(event: dict, vault: dict, amount: Decimal, threshold_units: Decimal) -> str: - """Build the Telegram message for one qualifying deposit.""" + """Build the Telegram message for one qualifying flow.""" chain_id = int(event["chainId"]) chain = Chain.from_chain_id(chain_id) explorer = EXPLORER_URLS.get(chain_id) vault_address = str(event["vaultAddress"]) tx_hash = str(event["transactionHash"]) tx = f"[{tx_hash}]({explorer}/tx/{tx_hash})" if explorer else tx_hash + flow_type = str(event["flow_type"]) lines = [ - "Small parent-vault deposit", + f"Small parent-vault {flow_type}", f"🏦 Vault: {address_link(vault_address, explorer)} ({vault['symbol']})", f"🪙 Amount: {format_amount(amount)} {vault['asset_symbol']}", f"📏 Threshold: < {format_amount(threshold_units)} {vault['asset_symbol']}", @@ -192,6 +204,9 @@ def build_alert_message(event: dict, vault: dict, amount: Decimal, threshold_uni f"👤 Owner: {address_link(str(event['owner']), explorer)}", f"💳 Sender: {address_link(str(event['sender']), explorer)}", ] + receiver = event.get("receiver") + if receiver: + lines.append(f"📥 Receiver: {address_link(str(receiver), explorer)}") transaction_from = event.get("transactionFrom") if transaction_from: lines.append(f"🚀 Tx From: {address_link(str(transaction_from), explorer)}") @@ -204,23 +219,31 @@ def cursor_from_event(event: dict) -> EventCursor: return EventCursor(int(event["blockNumber"]), int(event["logIndex"])) -def load_cursor(chain_id: int) -> EventCursor | None: - """Load a chain cursor from persistent monitor state.""" - raw = store.state_get(STATE_NAMESPACE, str(chain_id)) +def state_key(chain_id: int, flow_type: str) -> str: + """Return the persistent-state key for one chain and flow type.""" + if flow_type not in FLOW_ENTITY: + raise ValueError(f"Unknown flow type: {flow_type}") + return f"{chain_id}:{flow_type}" + + +def load_cursor(chain_id: int, flow_type: str) -> EventCursor | None: + """Load a chain/flow cursor from persistent monitor state.""" + key = state_key(chain_id, flow_type) + raw = store.state_get(STATE_NAMESPACE, key) if raw is None: return None try: payload = json.loads(raw) return EventCursor(int(payload["block_number"]), int(payload["log_index"])) except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: - raise RuntimeError(f"Invalid small-deposit cursor for chain {chain_id}: {raw}") from exc + raise RuntimeError(f"Invalid small-flow cursor for {key}: {raw}") from exc -def save_cursor(chain_id: int, cursor: EventCursor) -> None: - """Persist a successfully processed chain cursor.""" +def save_cursor(chain_id: int, flow_type: str, cursor: EventCursor) -> None: + """Persist a successfully processed chain/flow cursor.""" store.state_set( STATE_NAMESPACE, - str(chain_id), + state_key(chain_id, flow_type), json.dumps({"block_number": cursor.block_number, "log_index": cursor.log_index}), ) @@ -231,14 +254,14 @@ def process_event( threshold_units: Decimal, alert_sender: Callable[[Alert], None] = send_alert, ) -> bool: - """Evaluate one deposit, sending an alert when it is below the threshold.""" + """Evaluate one flow, sending an alert when it is below the threshold.""" vault_address = str(event["vaultAddress"]).lower() vault = vaults_by_address.get(vault_address) if vault is None: raise RuntimeError(f"Envio returned unknown parent vault {event['vaultAddress']}") decimals = int(vault["asset_decimals"]) - if not is_small_deposit(event["assets"], decimals, threshold_units): + if not is_small_flow(event["assets"], decimals, threshold_units): return False amount = format_units(event["assets"], decimals) @@ -247,34 +270,25 @@ def process_event( return True -def monitor_chain( - chain: Chain, +def monitor_flow_type( + chain_id: int, + flow_type: str, + addresses: list[str], + vaults_by_address: dict[str, dict], threshold_units: Decimal, lookback_seconds: int, page_size: int, now: int | None = None, ) -> tuple[int, int]: - """Fetch and process all new deposits for one chain.""" - vaults = fetch_kong_parent_vaults(chain) - if not vaults: - logger.warning("No active parent vaults returned for %s", chain.network_name) - return 0, 0 - - vaults_by_address = {str(vault["address"]).lower(): vault for vault in vaults} - # Envio stores checksummed addresses, while older rows or deployments may - # use lowercase. Supplying both forms keeps the case-sensitive filter safe. - addresses = sorted( - {address for vault in vaults for address in (str(vault["address"]), str(vault["address"]).lower())} - ) - - persisted_cursor = load_cursor(chain.chain_id) + """Fetch and process all new events of one type for a chain.""" + persisted_cursor = load_cursor(chain_id, flow_type) cursor = persisted_cursor or EventCursor(0, -1) since_ts = 0 if persisted_cursor else (now or int(time.time())) - lookback_seconds processed = 0 alerted = 0 while True: - events = load_deposits(chain.chain_id, addresses, cursor, since_ts, page_size) + events = load_events(flow_type, chain_id, addresses, cursor, since_ts, page_size) if events is None: break if not events: @@ -286,7 +300,7 @@ def monitor_chain( continue if process_event(event, vaults_by_address, threshold_units): alerted += 1 - save_cursor(chain.chain_id, event_cursor) + save_cursor(chain_id, flow_type, event_cursor) cursor = event_cursor processed += 1 @@ -296,6 +310,51 @@ def monitor_chain( return processed, alerted +def monitor_chain( + chain: Chain, + threshold_units: Decimal, + lookback_seconds: int, + page_size: int, + now: int | None = None, +) -> tuple[int, int]: + """Fetch and process deposits and withdrawals for one chain.""" + vaults = fetch_kong_parent_vaults(chain) + if not vaults: + logger.warning("No active parent vaults returned for %s", chain.network_name) + return 0, 0 + + vaults_by_address = {str(vault["address"]).lower(): vault for vault in vaults} + # Envio stores checksummed addresses, while older rows or deployments may + # use lowercase. Supplying both forms keeps the case-sensitive filter safe. + addresses = sorted( + {address for vault in vaults for address in (str(vault["address"]), str(vault["address"]).lower())} + ) + + processed = 0 + alerted = 0 + for flow_type in FLOW_TYPES: + flow_processed, flow_alerted = monitor_flow_type( + chain.chain_id, + flow_type, + addresses, + vaults_by_address, + threshold_units, + lookback_seconds, + page_size, + now, + ) + processed += flow_processed + alerted += flow_alerted + logger.info( + "%s %s: processed=%d alerted=%d", + chain.network_name, + flow_type, + flow_processed, + flow_alerted, + ) + return processed, alerted + + def parse_chain_ids(raw: str) -> list[Chain]: """Parse a comma-separated chain-ID list.""" chains: list[Chain] = [] @@ -306,10 +365,10 @@ def parse_chain_ids(raw: str) -> list[Chain]: def main() -> None: - """Run the small parent-vault deposit monitor.""" + """Run the small parent-vault flow monitor.""" default_chain_ids = ",".join(str(chain.chain_id) for chain in Chain) parser = argparse.ArgumentParser( - description="Alert on Yearn v3 parent-vault deposits below a token-unit threshold." + description="Alert on Yearn v3 parent-vault deposits and withdrawals below a token-unit threshold." ) parser.add_argument("--threshold-units", type=Decimal, default=DEFAULT_THRESHOLD_UNITS) parser.add_argument("--lookback-seconds", type=int, default=DEFAULT_LOOKBACK_SECONDS) diff --git a/tests/test_small_parent_deposits.py b/tests/test_small_parent_deposits.py deleted file mode 100644 index f69ce7ee..00000000 --- a/tests/test_small_parent_deposits.py +++ /dev/null @@ -1,116 +0,0 @@ -from decimal import Decimal - -from protocols.yearn import alert_small_parent_deposits as monitor -from utils.alert import AlertSeverity -from utils.chains import Chain - -VAULT = { - "address": "0xParent", - "name": "USDC yVault", - "symbol": "yvUSDC", - "decimals": 6, - "asset_address": "0xAsset", - "asset_symbol": "USDC", - "asset_decimals": 6, -} - - -def make_event(*, assets: str = "9999999999", block_number: int = 100, log_index: int = 2) -> dict: - return { - "id": f"1_{block_number}_{log_index}", - "vaultAddress": "0xParent", - "chainId": 1, - "blockNumber": block_number, - "blockTimestamp": 1_700_000_000, - "transactionHash": "0xTransaction", - "transactionFrom": "0xTransactionFrom", - "logIndex": log_index, - "sender": "0xSender", - "owner": "0xOwner", - "assets": assets, - "shares": assets, - } - - -def test_small_deposit_uses_normalized_token_units() -> None: - threshold = Decimal("10000") - - assert monitor.format_units("1234567", 6) == Decimal("1.234567") - assert monitor.is_small_deposit("9999999999", 6, threshold) - assert not monitor.is_small_deposit("10000000000", 6, threshold) - assert not monitor.is_small_deposit("10000000001", 6, threshold) - assert not monitor.is_small_deposit("0", 6, threshold) - - -def test_process_event_sends_low_alert_with_all_addresses() -> None: - alerts = [] - - did_alert = monitor.process_event( - make_event(), - {"0xparent": VAULT}, - Decimal("10000"), - alert_sender=alerts.append, - ) - - assert did_alert - assert len(alerts) == 1 - alert = alerts[0] - assert alert.severity is AlertSeverity.LOW - assert alert.protocol == "yearn" - assert "9,999.999999 USDC" in alert.message - assert "0xOwner" in alert.message - assert "0xSender" in alert.message - assert "0xTransactionFrom" in alert.message - assert "0xTransaction" in alert.message - - -def test_process_event_does_not_alert_at_threshold() -> None: - alerts = [] - - did_alert = monitor.process_event( - make_event(assets="10000000000"), - {"0xparent": VAULT}, - Decimal("10000"), - alert_sender=alerts.append, - ) - - assert not did_alert - assert alerts == [] - - -def test_monitor_chain_pages_and_persists_each_processed_event(monkeypatch) -> None: - first = make_event(block_number=100, log_index=2) - second = make_event(block_number=101, log_index=3) - calls = [] - saved = [] - - monkeypatch.setattr(monitor, "fetch_kong_parent_vaults", lambda _chain: [VAULT]) - monkeypatch.setattr(monitor, "load_cursor", lambda _chain_id: None) - monkeypatch.setattr(monitor, "save_cursor", lambda chain_id, cursor: saved.append((chain_id, cursor))) - monkeypatch.setattr(monitor, "process_event", lambda event, *_args: event is first) - - def fake_load(chain_id, addresses, cursor, since_ts, limit): - calls.append((chain_id, addresses, cursor, since_ts, limit)) - if len(calls) == 1: - return [first, second] - return [] - - monkeypatch.setattr(monitor, "load_deposits", fake_load) - - processed, alerted = monitor.monitor_chain( - Chain.MAINNET, - Decimal("10000"), - lookback_seconds=7200, - page_size=2, - now=1_700_010_000, - ) - - assert (processed, alerted) == (2, 1) - assert calls[0][0] == 1 - assert calls[0][2] == monitor.EventCursor(0, -1) - assert calls[0][3] == 1_700_002_800 - assert calls[1][2] == monitor.EventCursor(101, 3) - assert saved == [ - (1, monitor.EventCursor(100, 2)), - (1, monitor.EventCursor(101, 3)), - ] diff --git a/tests/test_small_parent_flows.py b/tests/test_small_parent_flows.py new file mode 100644 index 00000000..2525a441 --- /dev/null +++ b/tests/test_small_parent_flows.py @@ -0,0 +1,193 @@ +from decimal import Decimal + +from protocols.yearn import alert_small_parent_flows as monitor +from utils.alert import AlertSeverity +from utils.chains import Chain + +VAULT = { + "address": "0xParent", + "name": "USDC yVault", + "symbol": "yvUSDC", + "decimals": 6, + "asset_address": "0xAsset", + "asset_symbol": "USDC", + "asset_decimals": 6, +} + + +def make_event( + *, + flow_type: str = "deposit", + assets: str = "9999999999", + block_number: int = 100, + log_index: int = 2, +) -> dict: + event = { + "id": f"1_{block_number}_{log_index}", + "flow_type": flow_type, + "vaultAddress": "0xParent", + "chainId": 1, + "blockNumber": block_number, + "blockTimestamp": 1_700_000_000, + "transactionHash": "0xTransaction", + "transactionFrom": "0xTransactionFrom", + "logIndex": log_index, + "sender": "0xSender", + "owner": "0xOwner", + "assets": assets, + "shares": assets, + } + if flow_type == "withdrawal": + event["receiver"] = "0xReceiver" + return event + + +def test_small_flow_uses_normalized_token_units() -> None: + threshold = Decimal("10000") + + assert monitor.format_units("1234567", 6) == Decimal("1.234567") + assert monitor.is_small_flow("9999999999", 6, threshold) + assert not monitor.is_small_flow("10000000000", 6, threshold) + assert not monitor.is_small_flow("10000000001", 6, threshold) + assert not monitor.is_small_flow("0", 6, threshold) + + +def test_process_deposit_sends_low_alert_with_all_addresses() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(), + {"0xparent": VAULT}, + Decimal("10000"), + alert_sender=alerts.append, + ) + + assert did_alert + assert len(alerts) == 1 + alert = alerts[0] + assert alert.severity is AlertSeverity.LOW + assert alert.protocol == "yearn" + assert "Small parent-vault deposit" in alert.message + assert "9,999.999999 USDC" in alert.message + assert "0xOwner" in alert.message + assert "0xSender" in alert.message + assert "0xTransactionFrom" in alert.message + assert "0xTransaction" in alert.message + assert "Receiver" not in alert.message + + +def test_process_withdrawal_includes_asset_receiver() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(flow_type="withdrawal"), + {"0xparent": VAULT}, + Decimal("10000"), + alert_sender=alerts.append, + ) + + assert did_alert + assert len(alerts) == 1 + assert "Small parent-vault withdrawal" in alerts[0].message + assert "Receiver" in alerts[0].message + assert "0xReceiver" in alerts[0].message + + +def test_process_event_does_not_alert_at_threshold() -> None: + alerts = [] + + did_alert = monitor.process_event( + make_event(assets="10000000000"), + {"0xparent": VAULT}, + Decimal("10000"), + alert_sender=alerts.append, + ) + + assert not did_alert + assert alerts == [] + + +def test_load_events_selects_envio_entity_and_receiver(monkeypatch) -> None: + queries = [] + + def fake_gql(query, variables): + queries.append((query, variables)) + return {"data": {"events": [make_event(flow_type="withdrawal")]}} + + monkeypatch.setattr(monitor, "gql_request", fake_gql) + + events = monitor.load_events( + "withdrawal", + 1, + ["0xParent"], + monitor.EventCursor(10, 2), + 1_700_000_000, + 100, + ) + + assert "events: Withdraw(" in queries[0][0] + assert "receiver" in queries[0][0] + assert queries[0][1]["lastBlock"] == 10 + assert events and events[0]["flow_type"] == "withdrawal" + + +def test_monitor_flow_type_pages_and_persists_each_processed_event(monkeypatch) -> None: + first = make_event(flow_type="withdrawal", block_number=100, log_index=2) + second = make_event(flow_type="withdrawal", block_number=101, log_index=3) + calls = [] + saved = [] + + monkeypatch.setattr(monitor, "load_cursor", lambda _chain_id, _flow_type: None) + monkeypatch.setattr( + monitor, + "save_cursor", + lambda chain_id, flow_type, cursor: saved.append((chain_id, flow_type, cursor)), + ) + monkeypatch.setattr(monitor, "process_event", lambda event, *_args: event is first) + + def fake_load(flow_type, chain_id, addresses, cursor, since_ts, limit): + calls.append((flow_type, chain_id, addresses, cursor, since_ts, limit)) + if len(calls) == 1: + return [first, second] + return [] + + monkeypatch.setattr(monitor, "load_events", fake_load) + + processed, alerted = monitor.monitor_flow_type( + 1, + "withdrawal", + ["0xParent"], + {"0xparent": VAULT}, + Decimal("10000"), + lookback_seconds=7200, + page_size=2, + now=1_700_010_000, + ) + + assert (processed, alerted) == (2, 1) + assert calls[0][0] == "withdrawal" + assert calls[0][1] == 1 + assert calls[0][3] == monitor.EventCursor(0, -1) + assert calls[0][4] == 1_700_002_800 + assert calls[1][3] == monitor.EventCursor(101, 3) + assert saved == [ + (1, "withdrawal", monitor.EventCursor(100, 2)), + (1, "withdrawal", monitor.EventCursor(101, 3)), + ] + + +def test_monitor_chain_runs_deposit_and_withdrawal_streams(monkeypatch) -> None: + flow_types = [] + + monkeypatch.setattr(monitor, "fetch_kong_parent_vaults", lambda _chain: [VAULT]) + + def fake_monitor(_chain_id, flow_type, *_args): + flow_types.append(flow_type) + return 1, 1 + + monkeypatch.setattr(monitor, "monitor_flow_type", fake_monitor) + + result = monitor.monitor_chain(Chain.MAINNET, Decimal("10000"), 7200, 1000) + + assert result == (2, 2) + assert flow_types == ["deposit", "withdrawal"] From 0c6158c6588d552e801a11e2a2033ca9af6a6661 Mon Sep 17 00:00:00 2001 From: yvtapir Date: Fri, 28 Aug 2026 16:33:09 +0300 Subject: [PATCH 3/3] fix(yearn): compare small flows in raw asset units --- monitoring.yaml | 2 +- protocols/yearn/README.md | 6 +-- protocols/yearn/alert_small_parent_flows.py | 51 ++++++++++++--------- tests/test_small_parent_flows.py | 30 ++++++------ 4 files changed, 48 insertions(+), 41 deletions(-) diff --git a/monitoring.yaml b/monitoring.yaml index 578d8b7f..3bbeb100 100644 --- a/monitoring.yaml +++ b/monitoring.yaml @@ -472,7 +472,7 @@ protocols: - name: "Large Flows" description: "Deposit/withdrawal flows >=$500k USD (Katana withdrawals >=$50k; or 10% of vault totalSupply fallback for unpriced tokens)" - name: "Small Parent Vault Flows" - description: "Individual deposits and withdrawals below 10,000 normalized underlying-token units for active Yearn v3 parent vaults" + description: "Individual deposits and withdrawals with a raw ERC-4626 assets value below 10,000 for active Yearn v3 parent vaults" - name: "Timelock Delay" description: "Yearn TimelockController getMinDelay() < 7 days across Mainnet, Base, Arbitrum, Polygon, Optimism, Katana" - name: "Timelock Events" diff --git a/protocols/yearn/README.md b/protocols/yearn/README.md index 15278372..f99a444c 100644 --- a/protocols/yearn/README.md +++ b/protocols/yearn/README.md @@ -57,13 +57,13 @@ Optional flags: ## Small Parent Vault Flows -The script `yearn/alert_small_parent_flows.py` alerts on every positive deposit or withdrawal strictly below 10,000 normalized underlying-token units for an active Yearn v3 parent vault. The comparison is in token units, not USD: for example, both 9,999 USDC and 9,999 WETH qualify. +The script `yearn/alert_small_parent_flows.py` alerts on every deposit or withdrawal whose raw ERC-4626 `assets` value is strictly between 0 and 10,000 for an active Yearn v3 parent vault. The comparison happens before decimal normalization: 10,000 raw units equals 0.01 USDC, 0.0001 WBTC, or 0.00000000000001 WETH. ### Data Sources - **Parent vault discovery**: Kong GraphQL, filtered to Yearn v3 `vaultType: 1` vaults and excluding retired or hidden entries. - **Flow events**: Envio `Deposit` and `Withdraw` entities. Alerts include the ERC-4626 owner and sender, the transaction initiator, and the asset receiver for withdrawals. -- **Token decimals**: the parent vault's underlying asset metadata from Kong. +- **Token decimals**: the parent vault's underlying asset metadata from Kong, used only to show a human-readable amount alongside the raw value. Deposits and withdrawals are processed with independent per-chain `(blockNumber, logIndex)` cursors stored in the monitoring database. A cursor advances only after an event is successfully evaluated and, when applicable, delivered to Telegram. A new deployment starts each stream with a two-hour lookback. @@ -75,7 +75,7 @@ uv run protocols/yearn/alert_small_parent_flows.py Optional flags: -- `--threshold-units` (default: `10000`) +- `--threshold-raw` (default: `10000`) - `--lookback-seconds` (default: `7200`, used only before a chain cursor exists) - `--page-size` (default: `1000`) - `--chain-ids` (default: `1,10,8453,42161,137,747474`) diff --git a/protocols/yearn/alert_small_parent_flows.py b/protocols/yearn/alert_small_parent_flows.py index 4a685cb5..3eb4e791 100644 --- a/protocols/yearn/alert_small_parent_flows.py +++ b/protocols/yearn/alert_small_parent_flows.py @@ -30,7 +30,7 @@ ENVIO_GRAPHQL_URL = os.getenv("ENVIO_GRAPHQL_URL") DEFAULT_LOG_LEVEL = os.getenv("SMALL_PARENT_FLOWS_LOG_LEVEL", "INFO") -DEFAULT_THRESHOLD_UNITS = Decimal("10000") +DEFAULT_THRESHOLD_RAW = 10_000 DEFAULT_LOOKBACK_SECONDS = 7200 DEFAULT_PAGE_SIZE = 1000 PROTOCOL = "yearn" @@ -166,10 +166,10 @@ def format_units(raw_assets: str | int, decimals: int) -> Decimal: return Decimal(str(raw_assets)) / (Decimal(10) ** decimals) -def is_small_flow(raw_assets: str | int, decimals: int, threshold_units: Decimal) -> bool: - """Return whether a positive flow is strictly below the unit threshold.""" - amount = format_units(raw_assets, decimals) - return Decimal(0) < amount < threshold_units +def is_small_flow(raw_assets: str | int, threshold_raw: int) -> bool: + """Return whether a positive raw asset amount is below the threshold.""" + amount_raw = int(str(raw_assets)) + return 0 < amount_raw < threshold_raw def format_amount(amount: Decimal) -> str: @@ -185,7 +185,13 @@ def address_link(address: str, explorer: str | None) -> str: return address -def build_alert_message(event: dict, vault: dict, amount: Decimal, threshold_units: Decimal) -> str: +def build_alert_message( + event: dict, + vault: dict, + raw_assets: int, + amount: Decimal, + threshold_raw: int, +) -> str: """Build the Telegram message for one qualifying flow.""" chain_id = int(event["chainId"]) chain = Chain.from_chain_id(chain_id) @@ -198,8 +204,9 @@ def build_alert_message(event: dict, vault: dict, amount: Decimal, threshold_uni lines = [ f"Small parent-vault {flow_type}", f"🏦 Vault: {address_link(vault_address, explorer)} ({vault['symbol']})", - f"🪙 Amount: {format_amount(amount)} {vault['asset_symbol']}", - f"📏 Threshold: < {format_amount(threshold_units)} {vault['asset_symbol']}", + f"🔢 Raw Assets: {raw_assets:,}", + f"🪙 Normalized: {format_amount(amount)} {vault['asset_symbol']}", + f"📏 Raw Threshold: < {threshold_raw:,}", f"⛓️ Chain: {chain.network_name}", f"👤 Owner: {address_link(str(event['owner']), explorer)}", f"💳 Sender: {address_link(str(event['sender']), explorer)}", @@ -251,7 +258,7 @@ def save_cursor(chain_id: int, flow_type: str, cursor: EventCursor) -> None: def process_event( event: dict, vaults_by_address: dict[str, dict], - threshold_units: Decimal, + threshold_raw: int, alert_sender: Callable[[Alert], None] = send_alert, ) -> bool: """Evaluate one flow, sending an alert when it is below the threshold.""" @@ -260,12 +267,12 @@ def process_event( if vault is None: raise RuntimeError(f"Envio returned unknown parent vault {event['vaultAddress']}") - decimals = int(vault["asset_decimals"]) - if not is_small_flow(event["assets"], decimals, threshold_units): + raw_assets = int(str(event["assets"])) + if not is_small_flow(raw_assets, threshold_raw): return False - amount = format_units(event["assets"], decimals) - message = build_alert_message(event, vault, amount, threshold_units) + amount = format_units(raw_assets, int(vault["asset_decimals"])) + message = build_alert_message(event, vault, raw_assets, amount, threshold_raw) alert_sender(Alert(AlertSeverity.LOW, message, PROTOCOL)) return True @@ -275,7 +282,7 @@ def monitor_flow_type( flow_type: str, addresses: list[str], vaults_by_address: dict[str, dict], - threshold_units: Decimal, + threshold_raw: int, lookback_seconds: int, page_size: int, now: int | None = None, @@ -298,7 +305,7 @@ def monitor_flow_type( event_cursor = cursor_from_event(event) if event_cursor <= cursor: continue - if process_event(event, vaults_by_address, threshold_units): + if process_event(event, vaults_by_address, threshold_raw): alerted += 1 save_cursor(chain_id, flow_type, event_cursor) cursor = event_cursor @@ -312,7 +319,7 @@ def monitor_flow_type( def monitor_chain( chain: Chain, - threshold_units: Decimal, + threshold_raw: int, lookback_seconds: int, page_size: int, now: int | None = None, @@ -338,7 +345,7 @@ def monitor_chain( flow_type, addresses, vaults_by_address, - threshold_units, + threshold_raw, lookback_seconds, page_size, now, @@ -368,9 +375,9 @@ def main() -> None: """Run the small parent-vault flow monitor.""" default_chain_ids = ",".join(str(chain.chain_id) for chain in Chain) parser = argparse.ArgumentParser( - description="Alert on Yearn v3 parent-vault deposits and withdrawals below a token-unit threshold." + description="Alert on Yearn v3 parent-vault deposits and withdrawals below a raw-assets threshold." ) - parser.add_argument("--threshold-units", type=Decimal, default=DEFAULT_THRESHOLD_UNITS) + parser.add_argument("--threshold-raw", type=int, default=DEFAULT_THRESHOLD_RAW) parser.add_argument("--lookback-seconds", type=int, default=DEFAULT_LOOKBACK_SECONDS) parser.add_argument("--page-size", type=int, default=DEFAULT_PAGE_SIZE) parser.add_argument("--chain-ids", default=default_chain_ids) @@ -382,8 +389,8 @@ def main() -> None: format="[%(name)s] %(levelname)s %(message)s", stream=sys.stderr, ) - if args.threshold_units <= 0: - parser.error("--threshold-units must be positive") + if args.threshold_raw <= 0: + parser.error("--threshold-raw must be positive") if args.lookback_seconds < 0: parser.error("--lookback-seconds must be non-negative") if args.page_size <= 0: @@ -394,7 +401,7 @@ def main() -> None: for chain in parse_chain_ids(args.chain_ids): processed, alerted = monitor_chain( chain, - args.threshold_units, + args.threshold_raw, args.lookback_seconds, args.page_size, ) diff --git a/tests/test_small_parent_flows.py b/tests/test_small_parent_flows.py index 2525a441..aadc0944 100644 --- a/tests/test_small_parent_flows.py +++ b/tests/test_small_parent_flows.py @@ -18,7 +18,7 @@ def make_event( *, flow_type: str = "deposit", - assets: str = "9999999999", + assets: str = "5000", block_number: int = 100, log_index: int = 2, ) -> dict: @@ -42,14 +42,12 @@ def make_event( return event -def test_small_flow_uses_normalized_token_units() -> None: - threshold = Decimal("10000") - +def test_small_flow_uses_raw_asset_units() -> None: assert monitor.format_units("1234567", 6) == Decimal("1.234567") - assert monitor.is_small_flow("9999999999", 6, threshold) - assert not monitor.is_small_flow("10000000000", 6, threshold) - assert not monitor.is_small_flow("10000000001", 6, threshold) - assert not monitor.is_small_flow("0", 6, threshold) + assert monitor.is_small_flow("9999", 10_000) + assert not monitor.is_small_flow("10000", 10_000) + assert not monitor.is_small_flow("10001", 10_000) + assert not monitor.is_small_flow("0", 10_000) def test_process_deposit_sends_low_alert_with_all_addresses() -> None: @@ -58,7 +56,7 @@ def test_process_deposit_sends_low_alert_with_all_addresses() -> None: did_alert = monitor.process_event( make_event(), {"0xparent": VAULT}, - Decimal("10000"), + 10_000, alert_sender=alerts.append, ) @@ -68,7 +66,9 @@ def test_process_deposit_sends_low_alert_with_all_addresses() -> None: assert alert.severity is AlertSeverity.LOW assert alert.protocol == "yearn" assert "Small parent-vault deposit" in alert.message - assert "9,999.999999 USDC" in alert.message + assert "Raw Assets: 5,000" in alert.message + assert "Normalized: 0.005 USDC" in alert.message + assert "Raw Threshold: < 10,000" in alert.message assert "0xOwner" in alert.message assert "0xSender" in alert.message assert "0xTransactionFrom" in alert.message @@ -82,7 +82,7 @@ def test_process_withdrawal_includes_asset_receiver() -> None: did_alert = monitor.process_event( make_event(flow_type="withdrawal"), {"0xparent": VAULT}, - Decimal("10000"), + 10_000, alert_sender=alerts.append, ) @@ -97,9 +97,9 @@ def test_process_event_does_not_alert_at_threshold() -> None: alerts = [] did_alert = monitor.process_event( - make_event(assets="10000000000"), + make_event(assets="10000"), {"0xparent": VAULT}, - Decimal("10000"), + 10_000, alert_sender=alerts.append, ) @@ -158,7 +158,7 @@ def fake_load(flow_type, chain_id, addresses, cursor, since_ts, limit): "withdrawal", ["0xParent"], {"0xparent": VAULT}, - Decimal("10000"), + 10_000, lookback_seconds=7200, page_size=2, now=1_700_010_000, @@ -187,7 +187,7 @@ def fake_monitor(_chain_id, flow_type, *_args): monkeypatch.setattr(monitor, "monitor_flow_type", fake_monitor) - result = monitor.monitor_chain(Chain.MAINNET, Decimal("10000"), 7200, 1000) + result = monitor.monitor_chain(Chain.MAINNET, 10_000, 7200, 1000) assert result == (2, 2) assert flow_types == ["deposit", "withdrawal"]