Skip to main content
SuperVaults uses a polling-first monitoring model. No webhook system is implemented yet. Use these intervals as a baseline:
DataEndpointInterval
Vault state (PPS, pause, supply)GET /api/v1/vaults/{chain_id}/{address}15s
Analytics overview (APY, TVL)GET .../analytics/overview60s
Risk metricsGET .../analytics/risk120s
Service healthGET /api/v1/vaults/{chain_id}/{vault}/services30s
Infrastructure healthGET /health30s

​
Key Metrics

​
PPS Staleness

def check_pps_staleness(vault):
    remaining = vault.get("remaining_staleness_time", 0)
    if remaining < 3600:  # <1 hour
        alert("WARNING: PPS staleness — less than 1 hour remaining")
    if remaining < 600:   # <10 minutes
        alert("CRITICAL: PPS about to expire")

​
Service Success Rate

def check_service_health(services):
    for svc in services:
        rate = svc["success_rate"]
        if rate < 0.80:
            alert(f"CRITICAL: {svc['name']} success rate {rate:.0%}")
        elif rate < 0.95:
            alert(f"WARNING: {svc['name']} success rate {rate:.0%}")

​
Redemption Queue Depth

def check_redemption_queue(executions):
    pending = [e for e in executions if e["status"] == "pending"]
    if len(pending) > 10:
        alert(f"WARNING: {len(pending)} pending redemptions in queue")

​
Full Monitoring Loop

import requests
import time
import logging

EREBOR_URL = "https://erebor.superform.xyz"
JWT = "<your-jwt>"
CHAIN_ID = 8453
VAULT = "0xYourVault..."

headers = {"Authorization": f"Bearer {JWT}"}
log = logging.getLogger(__name__)

def fetch_vault_state():
    r = requests.get(
        f"{EREBOR_URL}/api/v1/vaults/{CHAIN_ID}/{VAULT}",
        headers=headers, timeout=10
    )
    r.raise_for_status()
    return r.json()

def fetch_service_health():
    r = requests.get(
        f"{EREBOR_URL}/api/v1/vaults/{CHAIN_ID}/{VAULT}/services",
        headers=headers, timeout=10
    )
    r.raise_for_status()
    return r.json()

def alert(severity: str, message: str):
    """Replace with Slack/PagerDuty integration."""
    log.warning(f"[{severity}] {message}")

def run_checks():
    # Vault state
    try:
        vault = fetch_vault_state()

        if vault.get("is_paused"):
            alert("CRITICAL", f"Vault {VAULT} is paused on-chain")

        remaining = vault.get("remaining_staleness_time", 9999)
        if remaining < 600:
            alert("CRITICAL", f"PPS expires in {remaining}s")
        elif remaining < 3600:
            alert("WARNING", f"PPS expires in {remaining // 60}min")

        tvl = vault.get("tvl_usd", 0)
        log.info(f"TVL: ${tvl:,.0f} | PPS: {vault.get('current_pps')} | Staleness: {remaining}s")

    except Exception as e:
        alert("ERROR", f"Failed to fetch vault state: {e}")

    # Services
    try:
        services = fetch_service_health()
        for svc in services:
            rate = svc.get("success_rate", 1.0)
            if rate < 0.80:
                alert("CRITICAL", f"Service {svc['name']}: {rate:.0%} success rate")
            elif rate < 0.95:
                alert("WARNING", f"Service {svc['name']}: {rate:.0%} success rate")
    except Exception as e:
        alert("ERROR", f"Failed to fetch service health: {e}")

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    while True:
        run_checks()
        time.sleep(30)

​
Alert Integrations

Replace the alert() stub with your notification system:

​
Slack

import httpx

def send_slack(message: str):
    httpx.post(
        os.environ["SLACK_WEBHOOK_URL"],
        json={"text": message}
    )

​
PagerDuty

import httpx

def pagerduty_alert(summary: str, severity: str = "warning"):
    httpx.post(
        "https://events.pagerduty.com/v2/enqueue",
        json={
            "routing_key": os.environ["PAGERDUTY_ROUTING_KEY"],
            "event_action": "trigger",
            "payload": {
                "summary": summary,
                "severity": severity,
                "source": "supervaults-monitor"
            }
        }
    )