DeFi Yield Aggregator Skill#

概述#

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

text

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

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

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

---

核心概念#

术语含义APY年化收益率（含复利效应）APR年化收益率（不含复利）TVLTotal Value Locked，锁仓量Vault收益池，用户存入资产、协议自动操作Auto-compounding自动复利，收益自动再质押Delta Neutral三角中性，对冲风险策略Leverage / 杠杆借款放大收益（同时放大风险）ILImpermanent Loss，无常损失

---

主流收益聚合协议#

🥇 Yearn Finance（以太坊）#

项目信息官网yearn.finance链Ethereum、Arbitrum、Polygon、Fantom特点主动策略 Vault（yVaults），顶级安全性代币YFI（治理代币）APY 来源yearn.finance/vaults

🥈 Beefy Finance（多链）#

项目信息官网beefy.finance链50+ 条链（BSC/ETH/Polygon/Arbitrum等）特点多链覆盖，收益 Vault 数量最多代币BIFIAPY 来源beefy.finance/vaults

🥉 Pendle Finance（创新）#

项目信息官网pendle.finance链Ethereum、Arbitrum、BSC特点收益代币化（PT / YT），可锁定未来收益适合预期利率下降，想提前锁定高收益

🏅 Gamma Strategies（Gamma）#

项目信息官网gamma.xyz链Ethereum、Polygon特点Uniswap V3 流动性管理，专攻 LP 收益适合想做市但不想主动管理 LP 仓位

🏅 Stargate（跨链）#

项目信息官网stargate.finance链多链特点跨链稳定币桥 + 收益池适合稳定币跨链 + 收益双赢

其他常见协议#

协议特点Convex FinanceCVX 生态，专注 Curve 收益Aura FinanceBalancer 收益聚合Gearbox杠杆收益协议RokenLido rETH 质押收益增强Eigenlayer再质押（Restaking）收益

---

工具#

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. 输出对比表 + 建议

操作步骤：

text

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

场景二：收益策略分析#

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

稳定币（USDC/USDT/DAI）：

text

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

ETH 生态：

text

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

BTC 生态：

text

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

场景三：仓位追踪#

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

---

常用脚本#

脚本 1：APY 批量查询#

文件： scripts/apy_checker.py

python

#!/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

python

#!/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

python

#!/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}")

---

风险提示 ⚠️#

风险类型说明防范无常损失 (IL)流动性池中资产价格波动导致损失优先选单币池（无 IL）智能合约漏洞协议被攻击导致资金损失选择经过审计的知名协议\\ Rug Pull**流动性突撤导致亏损检查 TVL 和团队背景Gas 成本L2 收益高但波动大计算净收益（扣 Gas 后）预言机操纵价格数据被操纵选择有 TWAP 保护的协议流动性不足提取时池子没足够流动性选择高 TVL 协议

---

参考链接#

• 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