DEV Community

Paarthurnax
Paarthurnax

Posted on

DeFi Yield Monitoring with OpenClaw: Track APYs Without Trusting Anyone

#cryptocurrency #ai #python #defi

DeFi Yield Monitoring with OpenClaw: Track APYs Without Trusting Anyone

DeFi yield monitoring with OpenClaw means you never have to trust a yield aggregator's dashboard, take a protocol's advertised APY at face value, or manually check 20 platforms every morning. Your local AI agent pulls yield data directly from public APIs — no middleman, no inflated numbers, no hidden fees — and tells you where the best real yields are right now. This guide walks through building a complete DeFi yield monitor from scratch.

The Problem With DeFi Yield Dashboards

DeFi platforms have a conflict of interest: they profit from TVL. Higher advertised APY = more deposits = more protocol revenue. So APY figures on protocol dashboards can be:

  • Inflated by token emissions — A 40% APY that's 35% governance tokens and 5% real yield
  • Volatile and misleading — APY changes every block, but dashboards show 7-day or 30-day averages
  • Missing risk context — A 200% APY on a new protocol might have zero audits
  • Incomplete — Each protocol shows only its own yields, not competitors

The solution: pull raw yield data from neutral third-party APIs and apply your own filters. DefiLlama is the gold standard for this — it aggregates yield data from 200+ protocols with no promotional bias.

Why DefiLlama Is the Right Data Source

DefiLlama is the most trusted DeFi data aggregator:

  • Open source (you can verify how numbers are calculated)
  • No protocol sponsorships or paid placement
  • Updates yield data in near-real-time
  • Free API with no authentication required
  • Includes risk metadata: audit status, TVL, protocol age 
# DefiLlama API requires NO API key
import requests

BASE_URL = "https://yields.llama.fi"

def get_all_pools():
    """Fetch all yield pool data from DefiLlama."""
    response = requests.get(f"{BASE_URL}/pools")
    return response.json()["data"]

# Test it - returns thousands of pools
all_pools = get_all_pools()
print(f"Total pools available: {len(all_pools)}")
print(f"Sample pool: {all_pools[0]}")
Enter fullscreen mode Exit fullscreen mode

Building the Yield Monitor

Here's a complete yield monitoring system that filters, scores, and alerts on DeFi opportunities:

import requests
import json
from datetime import datetime
from pathlib import Path

YIELD_STATE_FILE = Path("yield_state.json")

def get_all_pools():
    """Fetch all yield data from DefiLlama (no API key needed)."""
    try:
        response = requests.get("https://yields.llama.fi/pools", timeout=30)
        return response.json()["data"]
    except Exception as e:
        print(f"Error fetching yield data: {e}")
        return []

def filter_pools(
    pools,
    min_apy=3.0,          # Minimum APY to consider
    max_apy=500.0,         # Filter out likely scams/errors
    min_tvl_usd=1_000_000, # Min $1M TVL (liquidity filter)
    stablecoins_only=True, # Focus on lower-risk stablecoin yields
    target_assets=None     # Specific assets to track
):
    """
    Filter pools to find legitimate yield opportunities.

    Filtering logic:
    - min_apy: Anything below this isn't worth the gas + risk
    - max_apy: Extremely high APY = unsustainable token emissions or rug risk
    - min_tvl: Low TVL = exit liquidity risk
    - stablecoins_only: Avoids impermanent loss from volatile pair pools
    """
    if target_assets is None:
        target_assets = ["USDC", "USDT", "DAI", "ETH", "stETH", "wBTC"]

    filtered = []

    for pool in pools:
        apy = pool.get("apy", 0) or 0
        tvl = pool.get("tvlUsd", 0) or 0
        symbol = pool.get("symbol", "")
        chain = pool.get("chain", "")
        project = pool.get("project", "")
        stablecoin = pool.get("stablecoin", False)

        # Basic filters
        if apy < min_apy or apy > max_apy:
            continue
        if tvl < min_tvl_usd:
            continue
        if stablecoins_only and not stablecoin:
            # Allow ETH/wBTC even if not stablecoin flag
            has_target_asset = any(asset.upper() in symbol.upper() for asset in ["ETH", "WBTC", "STETH"])
            if not has_target_asset:
                continue

        # Asset filter
        if target_assets:
            has_target = any(asset.upper() in symbol.upper() for asset in target_assets)
            if not has_target:
                continue

        filtered.append({
            "pool_id": pool.get("pool"),
            "project": project,
            "chain": chain,
            "symbol": symbol,
            "apy": apy,
            "apy_base": pool.get("apyBase", 0) or 0,       # Real yield
            "apy_reward": pool.get("apyReward", 0) or 0,   # Token emissions
            "tvl_usd": tvl,
            "stablecoin": stablecoin,
            "il_risk": pool.get("ilRisk", "unknown"),       # Impermanent loss risk
            "audits": pool.get("audits"),
            "audit_links": pool.get("auditLinks", []),
            "url": pool.get("url", ""),
        })

    # Sort by APY descending
    return sorted(filtered, key=lambda x: x["apy"], reverse=True)

def score_pool_risk(pool):
    """
    Score a pool's risk level (1-10, lower = safer).

    Factors:
    - Token emission % of total APY (higher = riskier)
    - TVL (higher TVL = lower exit risk)
    - IL risk
    - Audit status
    """
    risk_score = 0
    risk_factors = []

    # Token emission risk
    if pool["apy"] > 0:
        emission_pct = pool["apy_reward"] / pool["apy"]
        if emission_pct > 0.8:
            risk_score += 4
            risk_factors.append(f"High emission dependency ({emission_pct:.0%} from token rewards)")
        elif emission_pct > 0.5:
            risk_score += 2
            risk_factors.append(f"Moderate emission dependency ({emission_pct:.0%} from rewards)")
        else:
            risk_factors.append(f"Mostly base yield ({(1-emission_pct):.0%} organic)")

    # TVL risk
    tvl = pool["tvl_usd"]
    if tvl < 5_000_000:
        risk_score += 3
        risk_factors.append(f"Low TVL (${tvl/1e6:.1f}M) — exit liquidity risk")
    elif tvl < 50_000_000:
        risk_score += 1
        risk_factors.append(f"Moderate TVL (${tvl/1e6:.1f}M)")
    else:
        risk_factors.append(f"High TVL (${tvl/1e6:.0f}M) — good liquidity")

    # IL risk
    if pool.get("il_risk") in ["yes", "high"]:
        risk_score += 2
        risk_factors.append("Impermanent loss risk")

    # Audit status
    if pool.get("audits") == "0" or not pool.get("audit_links"):
        risk_score += 2
        risk_factors.append("No audits found")
    else:
        risk_factors.append("Audited")

    risk_level = min(risk_score, 10)
    return risk_level, risk_factors

def format_yield_report(top_pools, max_pools=10):
    """Format a readable yield report."""
    report = f"🌾 *DeFi Yield Report — {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}*\n\n"

    if not top_pools:
        report += "No pools found matching your criteria.\n"
        return report

    for i, pool in enumerate(top_pools[:max_pools], 1):
        risk_score, risk_factors = score_pool_risk(pool)
        risk_emoji = "🟢" if risk_score <= 3 else "🟡" if risk_score <= 6 else "🔴"

        report += (
            f"{i}. *{pool['project']}* ({pool['chain']})\n"
            f"   Symbol: {pool['symbol']}\n"
            f"   APY: {pool['apy']:.1f}% "
            f"(base: {pool['apy_base']:.1f}%, rewards: {pool['apy_reward']:.1f}%)\n"
            f"   TVL: ${pool['tvl_usd']/1e6:.1f}M\n"
            f"   Risk: {risk_emoji} {risk_score}/10\n\n"
        )

    report += "\n⚠️ *Not financial advice. Research all protocols before depositing funds.*"
    return report

def send_telegram(token, chat_id, message):
    """Send Telegram message."""
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    requests.post(url, json={
        "chat_id": chat_id,
        "text": message,
        "parse_mode": "Markdown"
    })

def run_yield_monitor(telegram_token, telegram_chat_id):
    """Main yield monitoring function."""
    print("Fetching DeFi yield data from DefiLlama...")
    all_pools = get_all_pools()
    print(f"Found {len(all_pools)} total pools")

    # Filter for stablecoin yields
    stable_pools = filter_pools(
        all_pools,
        min_apy=4.0,
        max_apy=50.0,          # Filter extreme APYs
        min_tvl_usd=5_000_000, # $5M+ TVL
        stablecoins_only=True,
        target_assets=["USDC", "USDT", "DAI"]
    )
    print(f"Filtered to {len(stable_pools)} quality stablecoin pools")

    # Filter for ETH yields
    eth_pools = filter_pools(
        all_pools,
        min_apy=3.0,
        max_apy=30.0,
        min_tvl_usd=50_000_000, # Higher bar for ETH
        stablecoins_only=False,
        target_assets=["ETH", "stETH", "wETH"]
    )

    # Build report
    report = format_yield_report(stable_pools[:5])
    report += "\n\n"
    report += "*Top ETH Yield Pools:*\n"
    for pool in eth_pools[:3]:
        risk_score, _ = score_pool_risk(pool)
        risk_emoji = "🟢" if risk_score <= 3 else "🟡" if risk_score <= 6 else "🔴"
        report += (
            f"• {pool['project']} ({pool['chain']}): "
            f"{pool['apy']:.1f}% APY, TVL ${pool['tvl_usd']/1e6:.0f}M {risk_emoji}\n"
        )

    send_telegram(telegram_token, telegram_chat_id, report)
    print("Yield report sent to Telegram")

    # Save state for change detection
    state = {
        "last_run": datetime.utcnow().isoformat(),
        "top_pools": stable_pools[:10]
    }
    YIELD_STATE_FILE.write_text(json.dumps(state, indent=2))
Enter fullscreen mode Exit fullscreen mode

Detecting New High-Yield Opportunities

The most valuable alerts are when a new pool appears with better yield than your current positions:

def check_new_opportunities(current_pools, min_improvement_pct=20):
    """Alert when new high-yield opportunities appear."""
    state_file = YIELD_STATE_FILE

    if not state_file.exists():
        print("No previous state. Run full yield monitor first.")
        return

    previous_state = json.loads(state_file.read_text())
    previous_pools = {p["pool_id"]: p for p in previous_state.get("top_pools", [])}

    # Find genuinely new pools (not in previous top list)
    for pool in current_pools[:20]:
        pool_id = pool["pool_id"]

        if pool_id not in previous_pools:
            # New pool appeared in top 20
            risk_score, risk_factors = score_pool_risk(pool)

            if risk_score <= 5:  # Only alert on reasonably safe pools
                message = (
                    f"🆕 *New Yield Opportunity Detected*\n\n"
                    f"Protocol: {pool['project']} ({pool['chain']})\n"
                    f"Asset: {pool['symbol']}\n"
                    f"APY: {pool['apy']:.1f}%\n"
                    f"TVL: ${pool['tvl_usd']/1e6:.1f}M\n"
                    f"Risk Score: {risk_score}/10\n\n"
                    f"Factors:\n" + "\n".join(f"  • {f}" for f in risk_factors[:3]) +
                    f"\n\n⚠️ Always DYOR before depositing."
                )
                print(f"New opportunity alert: {pool['project']} {pool['apy']:.1f}%")
                # send_telegram(TELEGRAM_TOKEN, TELEGRAM_CHAT_ID, message)
Enter fullscreen mode Exit fullscreen mode

Key Red Flags to Filter

Before acting on any yield opportunity, check:

  1. APY > 100% for stablecoins — Almost always unsustainable token emissions. The tokens will dump.
  2. TVL < $1M — Tiny TVL means you could be the exit liquidity for the founders.
  3. No audits — In 2026, an unaudited DeFi protocol is a bet on the developers' intentions.
  4. New protocol (<6 months) — Most rug pulls happen in the first 6 months.
  5. Anonymous team — Not always bad, but higher risk profile.
  6. APY dropped dramatically since yesterday — Early exit signal.

Running the Monitor Automatically 

import schedule
import time
import os

TELEGRAM_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID")

# Daily yield report at 7 AM
schedule.every().day.at("07:00").do(
    lambda: run_yield_monitor(TELEGRAM_TOKEN, TELEGRAM_CHAT_ID)
)

# Check for new opportunities every 6 hours
schedule.every(6).hours.do(
    lambda: check_new_opportunities(get_filtered_pools())
)

print("DeFi Yield Monitor running...")
while True:
    schedule.run_pending()
    time.sleep(60)
Enter fullscreen mode Exit fullscreen mode

Get the Pre-Built Skill

The OpenClaw Skills Hub includes the Yield Optimizer skill ($29) — a packaged version of this monitor with risk scoring, daily digests, and new opportunity alerts already implemented and tested.

Browse skills at: https://paarthurnax970-debug.github.io/cryptoclawskills/

Get the complete DeFi monitoring setup with the Home AI Agent Kit.

Disclaimer: DeFi protocols carry significant smart contract risk, liquidity risk, and market risk. This monitor is an informational tool only. Nothing here constitutes financial or investment advice. Always do extensive research before depositing funds in any DeFi protocol. You can lose all deposited funds.

Top comments (0)