DeFi Yield Monitoring with OpenClaw: Track APYs Without Trusting Anyone

DeFi yield farming promised passive income. What it delivered, for many people, was a crash course in smart contract risk, impermanent loss, and the particular pain of watching a "safe" 200% APY pool collapse to zero overnight.

The problem wasn't DeFi itself — it was trusting the wrong sources, moving too slowly when conditions changed, and lacking the tools to monitor dozens of protocols simultaneously.

This guide shows you how to build a DeFi yield monitor with OpenClaw that tracks APYs across multiple protocols, detects suspicious spikes (rug pull warning sign), and alerts you when genuine opportunities or risks emerge — all without trusting any single data source.

Not financial advice. Paper trading only.

---

Why "Without Trusting Anyone" Matters

DeFi data has a trust problem:

• Protocol dashboards can display incorrect data during exploits
• Aggregators (DeFiLlama, etc.) have data lags and sometimes incorrect values
• Yield farming sites are often paid to promote specific protocols
• Discord announcements are frequently compromised or manipulated

The solution isn't to avoid all these sources — it's to cross-validate them. When multiple independent sources agree, you can trust the data. When they disagree, that's a signal to investigate before depositing.

---

The DeFi Data Landscape (Free Sources)

DeFiLlama API

The most reliable free DeFi data source. No API key required. Covers thousands of protocols across all major chains.

import urllib.request
import json

def get_defi_llama_yields():
    """Fetch yield data from DeFiLlama."""
    url = "https://yields.llama.fi/pools"

    with urllib.request.urlopen(url, timeout=30) as resp:
        data = json.loads(resp.read())

    pools = data.get("data", [])
    return pools

# Example pool entry:
# {
#   "chain": "Ethereum",
#   "project": "aave-v3",
#   "symbol": "USDC",
#   "tvlUsd": 1234567890,
#   "apy": 4.21,
#   "apyBase": 3.81,
#   "apyReward": 0.40,
#   "stablecoin": true,
#   "ilRisk": "no",
#   "pool": "0x..."
# }

The Graph Protocol

Decentralized indexer for on-chain DeFi data. Free queries with subgraph API.

def query_uniswap_pools(min_tvl=1_000_000):
    """Query Uniswap V3 pools via The Graph."""
    url = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"

    query = """
    {
      pools(
        first: 20,
        orderBy: volumeUSD,
        orderDirection: desc,
        where: { totalValueLockedUSD_gt: "%d" }
      ) {
        id
        token0 { symbol }
        token1 { symbol }
        feeTier
        totalValueLockedUSD
        volumeUSD
        poolDayData(first: 7, orderBy: date, orderDirection: desc) {
          date
          volumeUSD
          feesUSD
        }
      }
    }
    """ % min_tvl

    payload = json.dumps({"query": query}).encode()
    req = urllib.request.Request(
        url, data=payload,
        headers={"Content-Type": "application/json"}
    )

    with urllib.request.urlopen(req, timeout=30) as resp:
        data = json.loads(resp.read())

    return data.get("data", {}).get("pools", [])

Aave V3 Direct API

Aave provides its own data API for lending rates:

def get_aave_rates(chain="ethereum"):
    """Get Aave V3 lending/borrowing rates."""
    url = f"https://aave-api-v2.aave.com/data/markets-data"

    with urllib.request.urlopen(url, timeout=30) as resp:
        data = json.loads(resp.read())

    reserves = []
    for market in data.get("proto", []):
        if market.get("market", "").lower() == chain:
            reserves.extend(market.get("reserves", []))

    return [{
        "symbol": r.get("symbol"),
        "supply_apy": float(r.get("liquidityRate", 0)) * 100,
        "borrow_apy": float(r.get("variableBorrowRate", 0)) * 100,
        "tvl_usd": float(r.get("totalLiquidityUSD", 0)),
    } for r in reserves]

---

Building the Yield Monitor

Filter for Interesting Opportunities

def filter_quality_pools(pools, min_tvl=5_000_000, max_apy=50, min_apy=3):
    """
    Filter pools to find legitimate yield opportunities.

    Heuristic: APY > 50% with real TVL is almost always unsustainable
    or a warning sign of a Ponzi/rug. Focus on boring, stable yield.
    """
    quality = []
    for pool in pools:
        tvl = pool.get("tvlUsd", 0)
        apy = pool.get("apy", 0)

        if tvl < min_tvl:
            continue  # Too small — high exit risk
        if apy > max_apy:
            continue  # Suspiciously high — skip unless you understand the source
        if apy < min_apy:
            continue  # Not worth the smart contract risk
        if pool.get("outlier"):
            continue  # DeFiLlama flags these

        quality.append(pool)

    return sorted(quality, key=lambda p: p.get("apy", 0), reverse=True)

Detect APY Spikes (Risk Signal)

A sudden APY spike often means one of:

• Token rewards were just added (could be real, investigate)
• TVL dropped suddenly (denominator effect, concerning)
• Someone is attracting deposits before a rug

import sqlite3
from datetime import datetime, timedelta

def track_apy_changes(pools, db_path="defi_yields.db"):
    """Track APY over time and detect suspicious spikes."""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS yield_history (
            id INTEGER PRIMARY KEY,
            timestamp TEXT,
            pool_id TEXT,
            protocol TEXT,
            symbol TEXT,
            chain TEXT,
            apy REAL,
            tvl_usd REAL
        )
    """)

    alerts = []

    for pool in pools:
        pool_id = pool.get("pool", "unknown")
        protocol = pool.get("project", "unknown")
        symbol = pool.get("symbol", "unknown")
        chain = pool.get("chain", "unknown")
        apy = pool.get("apy", 0)
        tvl = pool.get("tvlUsd", 0)

        # Get recent history
        history = conn.execute("""
            SELECT apy, tvl_usd, timestamp FROM yield_history
            WHERE pool_id = ? AND timestamp > ?
            ORDER BY timestamp DESC LIMIT 10
        """, (pool_id, (datetime.utcnow() - timedelta(days=7)).isoformat())).fetchall()

        if history:
            avg_apy = sum(h[0] for h in history) / len(history)
            avg_tvl = sum(h[1] for h in history) / len(history)

            # Alert on >50% APY change
            if avg_apy > 0 and abs(apy - avg_apy) / avg_apy > 0.5:
                direction = "UP" if apy > avg_apy else "DOWN"
                alerts.append({
                    "type": "apy_spike",
                    "protocol": protocol,
                    "symbol": symbol,
                    "chain": chain,
                    "old_apy": avg_apy,
                    "new_apy": apy,
                    "direction": direction,
                    "severity": "HIGH" if direction == "UP" and apy > 20 else "MEDIUM",
                })

            # Alert on TVL drop > 30% (liquidity leaving)
            if avg_tvl > 0 and (avg_tvl - tvl) / avg_tvl > 0.30:
                alerts.append({
                    "type": "tvl_drop",
                    "protocol": protocol,
                    "symbol": symbol,
                    "chain": chain,
                    "old_tvl": avg_tvl,
                    "new_tvl": tvl,
                    "drop_pct": (avg_tvl - tvl) / avg_tvl * 100,
                    "severity": "HIGH",
                })

        # Record current data
        conn.execute("""
            INSERT INTO yield_history (timestamp, pool_id, protocol, symbol, chain, apy, tvl_usd)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (datetime.utcnow().isoformat(), pool_id, protocol, symbol, chain, apy, tvl))

    conn.commit()
    return alerts

---

Cross-Validating Data Sources

Here's the "trust no one" principle in action. Compare the same pool's APY from multiple sources:

def cross_validate_apy(protocol, symbol, chain, llama_apy):
    """
    Cross-validate APY from multiple sources.
    Returns confidence level based on agreement.
    """
    sources = {"defillama": llama_apy}

    # If it's Aave, check their own API
    if protocol.startswith("aave"):
        aave_rates = get_aave_rates(chain)
        aave_pool = next((r for r in aave_rates if r["symbol"] == symbol), None)
        if aave_pool:
            sources["aave_direct"] = aave_pool["supply_apy"]

    if len(sources) < 2:
        return "LOW", sources  # Only one source

    values = list(sources.values())
    avg = sum(values) / len(values)
    max_deviation = max(abs(v - avg) / avg for v in values) if avg > 0 else 1

    if max_deviation < 0.05:  # <5% deviation between sources
        confidence = "HIGH"
    elif max_deviation < 0.15:  # <15% deviation
        confidence = "MEDIUM"
    else:
        confidence = "LOW"  # Sources disagree significantly — investigate

    return confidence, sources

---

Full Monitoring Pipeline

def run_defi_monitor():
    """Complete DeFi yield monitoring pipeline."""
    import os
    TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
    CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")

    print("Fetching DeFiLlama yield data...")
    all_pools = get_defi_llama_yields()
    print(f"Fetched {len(all_pools)} pools")

    # Filter to quality opportunities
    quality = filter_quality_pools(all_pools, min_tvl=5_000_000, max_apy=50, min_apy=3)
    print(f"Quality pools: {len(quality)}")

    # Track and detect changes
    alerts = track_apy_changes(quality)

    # Report top opportunities
    top_5 = quality[:5]

    if TOKEN and CHAT_ID:
        # Send summary
        lines = ["*DeFi Yield Monitor* 📊\n"]
        lines.append("*Top Opportunities (>$5M TVL, 3-50% APY):*")
        for pool in top_5:
            lines.append(
                f"  • {pool['project']} {pool['symbol']} ({pool['chain']}): "
                f"`{pool['apy']:.2f}%` APY, TVL: `${pool['tvlUsd']/1e6:.1f}M`"
            )

        if alerts:
            lines.append(f"\n⚠️ *{len(alerts)} Alerts:*")
            for alert in alerts[:3]:  # Max 3 in summary
                if alert["type"] == "apy_spike":
                    lines.append(
                        f"  {alert['severity']}: {alert['protocol']} {alert['symbol']} "
                        f"APY {alert['direction']}: {alert['old_apy']:.1f}% → {alert['new_apy']:.1f}%"
                    )
                elif alert["type"] == "tvl_drop":
                    lines.append(
                        f"  {alert['severity']}: {alert['protocol']} {alert['symbol']} "
                        f"TVL -${(alert['old_tvl']-alert['new_tvl'])/1e6:.1f}M ({alert['drop_pct']:.0f}% drop)"
                    )

        lines.append("\n_Not financial advice. Always DYOR._")

        # Send via Telegram
        from telegram_alerts import send_message
        send_message(TOKEN, CHAT_ID, "\n".join(lines))

    return quality, alerts

if __name__ == "__main__":
    pools, alerts = run_defi_monitor()
    print(f"Monitoring complete. {len(alerts)} alerts generated.")

---

What to Watch For (DeFi Risk Checklist)

Before depositing into any pool your monitor surfaces:

Green flags:

• Protocol has been audited (check Certik, Code4rena)
• TVL has been stable or growing over weeks
• APY is consistent with the risk profile (stablecoin lending = 3-8%, not 200%)
• Smart contract is at least 6 months old without exploits
• Multiple data sources agree on APY

Red flags:

• APY spikes >50% in 24 hours with no announced reason
• TVL dropped sharply in the last 48 hours
• Protocol is less than 30 days old
• Reward token is entirely new/illiquid
• "Team is anon" + "no audit" + "high APY" = walk away

---

Schedule the Monitor

Run the monitor every 6 hours via OpenClaw:

# Add to your OpenClaw heartbeat skill
def check_defi_yields():
    from defi_monitor import run_defi_monitor
    pools, alerts = run_defi_monitor()

    if alerts:
        return f"⚠️ {len(alerts)} DeFi alerts — check Telegram"
    return f"DeFi OK — {len(pools)} quality pools tracked"

---

Start Monitoring, Not Chasing

DeFi yield farming isn't dead — but the days of blindly chasing the highest APY are over. The survivors are the ones who monitor carefully, diversify cautiously, and exit quickly when signals turn negative.

The full DeFi monitoring toolkit — including multi-protocol tracking, risk scoring, and historical APY charts — is in the OpenClaw kit:

👉 OpenClaw Home AI Agent Kit — Full Setup Guide

Monitor everything. Trust nothing. Move carefully.

---

---

🛠️ Also check out CryptoClaw Skills Hub — browse and install crypto skills for your OpenClaw agent: https://paarthurnax970-debug.github.io/cryptoclawskills/

Not financial advice. Paper trading only. DeFi involves significant smart contract risk. Always do your own research before depositing funds.