DeFi Yield Aggregator Skill

概述

收益聚合器（Yield Aggregator）是 DeFi 中的自动理财协议，核心逻辑：

用户入金 → 协议自动分配到最高收益池子 → 收益自动复投 → 用户受益

相比手动收益耕种的优势：

• 自动复利（收益再投资，省 Gas）
• 策略由专业团队维护（Yearn 等）
• 智能路由（自动切换到收益最高的池子）

核心概念

主流收益聚合协议

🥇 Yearn Finance（以太坊）

🥈 Beefy Finance（多链）

🥉 Pendle Finance（创新）

🏅 Gamma Strategies（Gamma）

🏅 Stargate（跨链）

其他常见协议

工具

API / Web 抓取

• extract_content_from_websites — 抓取 Yearn / Beefy / Pendle 等官方 APY 数据
• batch_web_search — 搜索最新收益策略、协议动态

Python 脚本

• scripts/apy_checker.py — 批量查询多协议 APY
• scripts/position_tracker.py — 追踪钱包在各协议中的仓位和收益
• scripts/yield_optimizer.py — 收益对比分析，优化建议

工作流程

场景一：查询最佳 APY

1. 确认用户资产类型：稳定币 / ETH / BTC / 山寨币
2. 确认链偏好和风险承受能力
3. 查询各协议对应 Vault APY
4. 输出对比表 + 建议

操作步骤：

1. 使用 apy_checker.py 拉取 Yearn / Beefy / Gamma 等协议 APY
2. 对比稳定币收益（USDC/USDT/DAI）
3. 考虑 Gas 成本（以太坊主网 vs L2）
4. 提示无常损失风险和流动性限制
5. 给出最优策略建议

场景二：收益策略分析

根据用户持仓类型推荐策略：

稳定币（USDC/USDT/DAI）：

最优：Yearn yUSDC Vault / Beefy USDC Vault
APY 范围：3%~15%（根据市场情况）
风险：低（单币池，无 IL）
注意：Gas 成本要低于收益
推荐链：Arbitrum / Polygon（低 Gas）

ETH 生态：

方案A：直接质押 Lido → 4~5% APY（安全稳定）
方案B：stETH 在 Curve/Yearn 循环LP → 5~10% APY
方案C：Yearn ETH Vault → 自动策略
风险：ETH 波动 + IL

BTC 生态：

方案A：WBTC 存入 BitBTC -> 4~6% APY
方案B：Curve wBTC/renBTC 池
风险：BTC 波动

场景三：仓位追踪

1. 用户提供钱包地址
2. 调用 position_tracker.py 扫描
3. 输出各协议的资产分布 + 估算收益
4. 提示未领取奖励情况

常用脚本

脚本 1：APY 批量查询

文件：scripts/apy_checker.py

#!/usr/bin/env python3
"""
DeFi APY 批量查询器
查询 Yearn、Beefy、Pendle 等主流协议的 APY 数据
"""
import requests
import json
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class VaultInfo:
    protocol: str
    chain: str
    asset: str
    vault_address: str
    apy: float  # percentage
    tvl_usd: float
    has_reward_token: bool = False
    reward_apy: float = 0.0

def query_yearn_vaults() -> List[VaultInfo]:
    """查询 Yearn Finance Vaults"""
    url = "https://yearn.vision/vaults"
    results = []
    try:
        resp = requests.get(url, timeout=10)
        data = resp.json()
        for vault in data.get("data", [])[:20]:
            results.append(VaultInfo(
                protocol="Yearn",
                chain=vault.get("chain", "ethereum"),
                asset=vault.get("symbol", "").replace("yv", "").upper(),
                vault_address=vault.get("address", ""),
                apy=float(vault.get("apy", {}).get("net_apy", 0) or 0) * 100,
                tvl_usd=float(vault.get("tvl", {}).get("tvl", 0) or 0),
            ))
    except Exception as e:
        print(f"Yearn API 失败: {e}")
    return results

def query_beefy_vaults(asset: str = "") -> List[VaultInfo]:
    """查询 Beefy Finance Vaults"""
    url = "https://api.beefy.finance/vaults"
    results = []
    try:
        resp = requests.get(url, timeout=10)
        data = resp.json()
        for vault in data:
            if asset and asset.upper() not in vault.get("token", "").upper():
                continue
            results.append(VaultInfo(
                protocol="Beefy",
                chain=vault.get("chain", ""),
                asset=vault.get("token", ""),
                vault_address=vault.get("earnContractAddress", ""),
                apy=float(vault.get("apy", 0) or 0) * 100,
                tvl_usd=float(vault.get("tvl", 0) or 0),
            ))
    except Exception as e:
        print(f"Beefy API 失败: {e}")
    return results

def query_pendle_markets() -> List[VaultInfo]:
    """查询 Pendle Finance 市场"""
    url = "https://api.pendle.finance/vip/data/api/v1/markets"
    results = []
    headers = {"Accept": "application/json"}
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        data = resp.json().get("data", []) if isinstance(resp.json(), dict) else resp.json()
        for m in data[:10]:
            results.append(VaultInfo(
                protocol="Pendle",
                chain="ethereum",
                asset=m.get("underlyingAsset", ""),
                vault_address=m.get("address", ""),
                apy=float(m.get("apy", 0) or 0),
                tvl_usd=float(m.get("tvlUsd", 0) or 0),
            ))
    except Exception as e:
        print(f"Pendle API 失败: {e}")
    return results

def format_results(vaults: List[VaultInfo], top_n: int = 10):
    """格式化输出"""
    # 排序
    sorted_vaults = sorted(vaults, key=lambda x: x.apy, reverse=True)
    print(f"n{'协议':<10} {'链':<10} {'资产':<12} {'APY':>8} {'TVL($)':>14}")
    print("-" * 60)
    for v in sorted_vaults[:top_n]:
        apy_str = f"{v.apy:.2f}%" if v.apy > 0 else "N/A"
        tvl_str = f"${v.tvl_usd:,.0f}" if v.tvl_usd > 0 else "N/A"
        print(f"{v.protocol:<10} {v.chain:<10} {v.asset:<12} {apy_str:>8} {tvl_str:>14}")

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="DeFi APY 查询器")
    parser.add_argument("--asset", "-a", default="", help="过滤资产，如 USDC、ETH")
    parser.add_argument("--top", "-n", type=int, default=10, help="显示前 N 名")
    args = parser.parse_args()

    print("=" * 60)
    print("  DeFi APY 查询器")
    print("=" * 60)

    all_vaults = []
    all_vaults += query_yearn_vaults()
    all_vaults += query_beefy_vaults(args.asset)

    if all_vaults:
        format_results(all_vaults, args.top)
    else:
        print("未获取到数据，请检查网络")

脚本 2：仓位追踪器

文件：scripts/position_tracker.py

#!/usr/bin/env python3
"""
DeFi 仓位追踪器
查询钱包在各收益协议中的持仓和累计收益

用法:
  export WALLET=0xYourAddress
  python3 scripts/position_tracker.py
"""
import os
import requests
from web3 import Web3
from dataclasses import dataclass
from typing import Dict, List

WALLET = os.getenv("WALLET", "0x0000000000000000000000000000000000000000")

# 常用协议 Vault 地址（部分）
KNOWN_VAULTS = {
    "yearn": {
        "name": "Yearn Finance",
        "vaults": {
            "0x5f18C75AbDAe578b477E319BcA4F55eFEF2eF892": "yUSDC",
            "0xa354F3587Ae0fB13C52e4873BCCC4767C8eFc2D9": "yUSDT",
            "0x8d0C6B68D3E1D2d8f2d0D3F9d5E3f2E8D2E1E0E1": "yDAI",
        }
    },
    "beefy": {
        "name": "Beefy Finance",
        "vaults": {
            "0x0000000000000000000000000000000000000000": "beUSDC",
        }
    }
}

@dataclass
class Position:
    protocol: str
    vault: str
    asset: str
    balance: float
    value_usd: float
    apy: float
    pending_rewards: float = 0.0

def get_token_balance(w3: Web3, token_addr: str, wallet: str) -> float:
    """查询 Token 余额"""
    erc20_abi = [
        {
            "inputs": [{"name": "account", "type": "address"}],
            "name": "balanceOf",
            "outputs": [{"name": "", "type": "uint256"}],
            "stateMutability": "view",
            "type": "function"
        },
        {
            "inputs": [],
            "name": "decimals",
            "outputs": [{"name": "", "type": "uint8"}],
            "stateMutability": "view",
            "type": "function"
        },
        {
            "inputs": [],
            "name": "symbol",
            "outputs": [{"name": "", "type": "string"}],
            "stateMutability": "view",
            "type": "function"
        }
    ]
    try:
        contract = w3.eth.contract(address=Web3.to_checksum_address(token_addr), abi=erc20_abi)
        bal = contract.functions.balanceOf(Web3.to_checksum_address(wallet)).call()
        decimals = contract.functions.decimals().call()
        return bal / (10 ** decimals)
    except Exception:
        return 0.0

def estimate_yearn_position(vault_addr: str, wallet: str, rpc_url: str) -> Optional[Position]:
    """估算 Yearn Vault 仓位"""
    w3 = Web3(Web3.HTTPProvider(rpc_url))
    if not w3.is_connected():
        return None
    balance = get_token_balance(w3, vault_addr, wallet)
    if balance <= 0:
        return None
    return Position(
        protocol="Yearn",
        vault=vault_addr[:10] + "...",
        asset="Vault Share",
        balance=balance,
        value_usd=balance * 1.0,  # 简化估算
        apy=0.0,
    )

def main():
    if WALLET.startswith("0x000"):
        print("⚠️ 请设置钱包地址:")
        print("   export WALLET=0xYourAddress")
        print("   python3 scripts/position_tracker.py")
        return

    print(f"🔍 扫描地址: {WALLET}")
    print("-" * 50)

    rpc = "https://eth.llamarpc.com"
    positions: List[Position] = []

    # 扫描 Yearn
    print("n📊 Yearn Vaults...")
    for addr, name in KNOWN_VAULTS["yearn"]["vaults"].items():
        pos = estimate_yearn_position(addr, WALLET, rpc)
        if pos:
            positions.append(pos)
            print(f"   ✅ {name}: {pos.balance:.4f} shares (≈${pos.value_usd:.2f})")

    if not positions:
        print("   📭 未发现已知的 Yearn 仓位")

    print("n📊 Beefy Finance...")
    print("   (需要 Beefy API，建议直接在 beefy.finance 钱包连接查看)")

    print("n" + "=" * 50)
    total_value = sum(p.value_usd for p in positions)
    total_apy = sum(p.apy * p.value_usd for p in positions) / max(total_value, 1)
    print(f"n💰 累计 TVL: ${total_value:,.2f}")
    print(f"📈 加权 APY: {total_apy:.2f}%")

if __name__ == "__main__":
    main()

脚本 3：收益优化建议器

文件：scripts/yield_optimizer.py

#!/usr/bin/env python3
"""
DeFi 收益优化建议器
根据用户资产和风险偏好，推荐最佳收益策略
"""
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Strategy:
    protocol: str
    vault_name: str
    asset: str
    chain: str
    apy: float
    min_deposit: float
    risk_level: str  # low / medium / high
    il_risk: bool
    lockup: bool
    notes: str

STRATEGIES = [
    # 稳定币
    Strategy("Yearn", "yUSDC", "USDC", "Ethereum", 4.5, 0, "low", False, False, "安全首选，TVL 高"),
    Strategy("Yearn", "yUSDC", "USDC", "Arbitrum", 6.2, 0, "low", False, False, "L2 低 Gas，APY 更高"),
    Strategy("Beefy", "USDC", "USDC", "Polygon", 5.8, 0, "low", False, False, "Polygon 低费用"),
    Strategy("Beefy", "USDT", "USDT", "Arbitrum", 7.1, 0, "medium", False, False, "APY 较高"),
    Strategy("Curve", "3pool", "USDT/USDC/DAI", "Ethereum", 3.2, 1000, "low", False, False, "主流稳定币池"),
    Strategy("Convex", "3pool", "USDC", "Ethereum", 4.8, 10, "low", False, False, "Curve + CVX 叠加收益"),
    # ETH
    Strategy("Lido", "stETH", "ETH", "Ethereum", 4.3, 0, "low", False, False, "最安全的 ETH 质押"),
    Strategy("Yearn", "yETH", "ETH", "Ethereum", 5.1, 0, "medium", True, False, "自动策略，有 IL"),
    Strategy("Rocket Pool", "rETH", "ETH", "Ethereum", 4.6, 0.1, "low", False, False, "去中心化，比 Lido 收益略高"),
    Strategy("Gearbox", "dGEAR", "ETH", "Ethereum", 15.0, 0.5, "high", True, True, "杠杆策略，风险高"),
    # BTC
    Strategy("Curve", "renBTC/wBTC", "WBTC", "Ethereum", 2.1, 0.01, "medium", True, False, "BTC 稳定对，有 IL"),
    Strategy("Yearn", "yWBTC", "WBTC", "Arbitrum", 4.8, 0, "medium", True, False, "Yearn 托管，省 Gas"),
]

def optimize(asset: str, risk: str = "low", min_apy: float = 0) -> List[Strategy]:
    """根据条件过滤策略"""
    results = []
    for s in STRATEGIES:
        if asset.lower() not in s.asset.lower():
            continue
        if risk == "low" and s.risk_level not in ("low",):
            continue
        if risk == "medium" and s.risk_level == "high":
            continue
        if s.apy < min_apy:
            continue
        results.append(s)
    return sorted(results, key=lambda x: x.apy, reverse=True)

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="DeFi 收益优化建议")
    parser.add_argument("--asset", "-a", default="USDC", help="资产类型")
    parser.add_argument("--risk", "-r", default="low", choices=["low", "medium", "high"], help="风险等级")
    parser.add_argument("--min-apy", "-m", type=float, default=0, help="最低 APY 要求")
    args = parser.parse_args()

    print("=" * 60)
    print(f"  收益优化建议 | 资产: {args.asset} | 风险: {args.risk}")
    print("=" * 60)

    results = optimize(args.asset, args.risk, args.min_apy)
    if not results:
        print("n未找到符合条件的策略")
    for s in results[:5]:
        print(f"n🏦 {s.protocol} - {s.vault_name}")
        print(f"   资产: {s.asset} | 链: {s.chain}")
        print(f"   APY: {s.apy:.2f}% | 风险: {s.risk_level.upper()}")
        print(f"   IL 风险: {'⚠️ 有' if s.il_risk else '✅ 无'} | 锁仓: {'⏱️ 有' if s.lockup else '✅ 无'}")
        print(f"   备注: {s.notes}")

风险提示 ⚠️

参考链接

• Yearn Vaults：https://yearn.finance/vaults
• Beefy Vaults：https://beefy.finance/vaults
• Pendle：https://pendle.finance
• Gamma Strategies：https://gamma.xyz
• DeFi Llama（收益对比）：https://defillama.com/yields
• Dune Analytics（收益看板）：https://dune.com