DeFi Yield Farming for AI Agents

Yield farming — deploying capital into DeFi protocols to earn trading fees, liquidity mining rewards, and lending interest — is one of the highest-leverage activities available to autonomous AI agents. An agent never sleeps, never forgets to compound, and can monitor hundreds of pools simultaneously. This guide covers the mechanics, Python tooling, impermanent loss management, and integration with the Purple Flea wallet API.

The DeFi Yield Landscape

Yield sources in DeFi fall into three main categories, each with different risk profiles and capital efficiency requirements. Agents should allocate across all three rather than concentrating in any single strategy.

Fetching Yields Programmatically

Multiple aggregator APIs expose DeFi APY data in structured form. The defillama API is free, comprehensive, and reliable.

import httpx
import asyncio
from dataclasses import dataclass

@dataclass
class YieldPool:
    pool_id: str
    protocol: str
    chain: str
    symbol: str
    tvl_usd: float
    apy: float        # total APY including rewards
    apy_base: float   # fee APY only (no token emissions)
    il_risk: str      # "none", "low", "medium", "high"
    stablecoin: bool

async def fetch_defi_yields(
    min_tvl: float = 1_000_000,  # minimum $1M TVL
    min_apy: float = 3.0,
    chains: list[str] | None = None,
) -> list[YieldPool]:
    """Fetch yield opportunities from DeFi Llama."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://yields.llama.fi/pools",
            timeout=30.0
        )
        data = resp.json()

    pools = []
    for p in data.get("data", []):
        if p.get("tvlUsd", 0) < min_tvl:
            continue
        apy = p.get("apy") or 0
        if apy < min_apy:
            continue
        if chains and p.get("chain", "").lower() not in [c.lower() for c in chains]:
            continue
        pools.append(YieldPool(
            pool_id=p.get("pool", ""),
            protocol=p.get("project", ""),
            chain=p.get("chain", ""),
            symbol=p.get("symbol", ""),
            tvl_usd=p.get("tvlUsd", 0),
            apy=apy,
            apy_base=p.get("apyBase") or 0,
            il_risk=p.get("ilRisk", "unknown"),
            stablecoin=p.get("stablecoin", False),
        ))

    # Sort by risk-adjusted APY: penalize non-stable pairs
    def risk_adj_apy(pool: YieldPool) -> float:
        base = pool.apy
        if pool.il_risk == "high": base *= 0.5
        elif pool.il_risk == "medium": base *= 0.75
        return base

    return sorted(pools, key=risk_adj_apy, reverse=True)

# Usage
async def main():
    pools = await fetch_defi_yields(min_tvl=5_000_000, chains=["Ethereum", "Arbitrum"])
    for p in pools[:10]:
        print(f"{p.protocol}/{p.symbol}: {p.apy:.1f}% APY | IL: {p.il_risk} | TVL: ${p.tvl_usd/1e6:.1f}M")

asyncio.run(main())

Understanding and Mitigating Impermanent Loss

Impermanent loss (IL) is the opportunity cost of providing liquidity compared to simply holding the assets. It occurs whenever the price ratio of the two pooled assets diverges from the ratio at deposit time.

IL formula

For a 50/50 pool with price ratio change factor r:

import math

def impermanent_loss(price_ratio_change: float) -> float:
    """
    Compute impermanent loss as a fraction of hold value.
    price_ratio_change: new_price / initial_price (e.g., 2.0 = price doubled)

    Returns: loss as negative fraction (e.g., -0.057 = 5.7% loss vs holding)
    """
    r = price_ratio_change
    lp_value = 2 * math.sqrt(r) / (1 + r)
    return lp_value - 1.0  # negative = loss vs holding

# Examples
print(f"2x price change: {impermanent_loss(2.0):.2%}")    # -5.7%
print(f"3x price change: {impermanent_loss(3.0):.2%}")    # -13.4%
print(f"5x price change: {impermanent_loss(5.0):.2%}")    # -25.5%
print(f"0.5x price change: {impermanent_loss(0.5):.2%}")  # -5.7% (symmetric)

IL mitigation strategies for agents

Uniswap v3 Concentrated Liquidity

Uniswap v3 allows liquidity providers to specify a price range. Capital only earns fees when the current price is within the range. This dramatically increases capital efficiency but requires active management — when price exits the range, fees stop accruing and the position becomes 100% of the weaker asset.

from web3 import Web3
from eth_account import Account
import json

# Uniswap v3 Position Manager ABI (simplified)
POSITION_MANAGER_ABI = json.loads("""[
  {"inputs":[{"components":[
    {"name":"token0","type":"address"},
    {"name":"token1","type":"address"},
    {"name":"fee","type":"uint24"},
    {"name":"tickLower","type":"int24"},
    {"name":"tickUpper","type":"int24"},
    {"name":"amount0Desired","type":"uint256"},
    {"name":"amount1Desired","type":"uint256"},
    {"name":"amount0Min","type":"uint256"},
    {"name":"amount1Min","type":"uint256"},
    {"name":"recipient","type":"address"},
    {"name":"deadline","type":"uint256"}
  ],"name":"params","type":"tuple"}],
  "name":"mint","outputs":[
    {"name":"tokenId","type":"uint256"},
    {"name":"liquidity","type":"uint128"},
    {"name":"amount0","type":"uint256"},
    {"name":"amount1","type":"uint256"}
  ],"stateMutability":"payable","type":"function"}
]""")

UNISWAP_V3_POSITION_MANAGER = "0xC36442b4a4522E871399CD717aBDD847Ab11FE88"

def price_to_tick(price: float, tick_spacing: int = 60) -> int:
    """Convert a price to the nearest valid Uniswap v3 tick."""
    import math
    raw_tick = math.floor(math.log(price) / math.log(1.0001))
    return (raw_tick // tick_spacing) * tick_spacing

async def open_uniswap_position(
    w3: Web3,
    account: Account,
    token0: str,
    token1: str,
    fee_tier: int,       # 500 = 0.05%, 3000 = 0.3%, 10000 = 1%
    current_price: float,
    range_pct: float,    # e.g., 0.10 = ±10% range
    amount0: int,        # in token0 wei
    amount1: int,        # in token1 wei
) -> dict:
    """Open a concentrated liquidity position on Uniswap v3."""
    tick_spacing = {500: 10, 3000: 60, 10000: 200}[fee_tier]
    lower_price = current_price * (1 - range_pct)
    upper_price = current_price * (1 + range_pct)
    tick_lower = price_to_tick(lower_price, tick_spacing)
    tick_upper = price_to_tick(upper_price, tick_spacing)

    contract = w3.eth.contract(
        address=UNISWAP_V3_POSITION_MANAGER,
        abi=POSITION_MANAGER_ABI
    )
    import time
    deadline = int(time.time()) + 1200  # 20 min
    slippage = 0.005  # 0.5%

    tx = contract.functions.mint({
        "token0": Web3.to_checksum_address(token0),
        "token1": Web3.to_checksum_address(token1),
        "fee": fee_tier,
        "tickLower": tick_lower,
        "tickUpper": tick_upper,
        "amount0Desired": amount0,
        "amount1Desired": amount1,
        "amount0Min": int(amount0 * (1 - slippage)),
        "amount1Min": int(amount1 * (1 - slippage)),
        "recipient": account.address,
        "deadline": deadline,
    }).build_transaction({
        "from": account.address,
        "nonce": w3.eth.get_transaction_count(account.address),
        "gas": 500_000,
    })
    signed = account.sign_transaction(tx)
    tx_hash = w3.eth.send_raw_transaction(signed.rawTransaction)
    receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
    return {"tx_hash": tx_hash.hex(), "status": receipt.status}

Auto-Compounding Strategy

The most powerful feature of an autonomous agent is compounding. A human farmer checks yields weekly at best; an agent can compound every hour, dramatically increasing effective APY through the power of frequent reinvestment.

Compound frequency vs. gas cost analysis

def optimal_compound_frequency(
    principal_usd: float,
    apy_pct: float,
    gas_cost_usd: float,
    holding_days: int = 365,
) -> dict:
    """
    Find the compound frequency that maximizes net returns
    after accounting for gas costs.
    """
    import math
    results = {}
    r = apy_pct / 100
    for compounds_per_year in [1, 4, 12, 52, 365, 8760]:  # annual to hourly
        period_rate = r / compounds_per_year
        final_value = principal_usd * (1 + period_rate) ** (
            compounds_per_year * holding_days / 365
        )
        total_gas = gas_cost_usd * compounds_per_year * (holding_days / 365)
        net_gain = final_value - principal_usd - total_gas
        results[compounds_per_year] = {
            "label": {1:"Annual",4:"Quarterly",12:"Monthly",
                      52:"Weekly",365:"Daily",8760:"Hourly"}[compounds_per_year],
            "final_value": final_value,
            "total_gas": total_gas,
            "net_gain": net_gain,
            "effective_apy": (final_value - total_gas) / principal_usd - 1,
        }
    # Find optimal
    best = max(results.items(), key=lambda x: x[1]["net_gain"])
    return {"optimal_frequency": best[1]["label"], "details": results}

# $10,000 at 15% APY, $3 gas per compound
result = optimal_compound_frequency(10_000, 15, 3.0)
print(f"Optimal frequency: {result['optimal_frequency']}")
# Typically: Daily compounding wins at this scale; hourly wins above ~$50K principal

Aave Lending Integration

Aave is the most battle-tested lending protocol with audited contracts and deep liquidity. Agents can deposit assets, earn the supply APY, and optionally borrow against the deposit to create leveraged yield positions.

import httpx

AAVE_V3_SUBGRAPH = "https://api.thegraph.com/subgraphs/name/aave/protocol-v3"

async def fetch_aave_rates() -> list[dict]:
    """Fetch current supply and borrow APY from Aave v3 via subgraph."""
    query = """
    {
      reserves(first: 20, orderBy: totalLiquidityAsToken, orderDirection: desc) {
        symbol
        liquidityRate
        variableBorrowRate
        totalDeposits
        utilizationRate
        isActive
      }
    }
    """
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            AAVE_V3_SUBGRAPH,
            json={"query": query},
            timeout=15.0
        )
        data = resp.json()

    reserves = []
    for r in data.get("data", {}).get("reserves", []):
        if not r["isActive"]:
            continue
        # Aave rates are in ray units (1e27); convert to APY
        supply_apy = (float(r["liquidityRate"]) / 1e27) * 100
        borrow_apy = (float(r["variableBorrowRate"]) / 1e27) * 100
        reserves.append({
            "symbol": r["symbol"],
            "supply_apy": supply_apy,
            "borrow_apy": borrow_apy,
            "utilization": float(r.get("utilizationRate", 0)),
        })
    return sorted(reserves, key=lambda x: x["supply_apy"], reverse=True)

# Leveraged yield: deposit USDC, borrow USDT, deposit USDT
# Net yield = supply_APY(USDC) + supply_APY(USDT) - borrow_APY(USDT)
# With 75% LTV and 2x leverage ≈ 1.75x base yield

Purple Flea Wallet Integration

Purple Flea's wallet API provides agents with a managed multi-chain wallet that handles key management, nonce tracking, and gas estimation. This is the recommended approach for agents that should not manage raw private keys.

import httpx
from typing import Literal

PURPLE_FLEA_API = "https://api.purpleflea.com"

class YieldFarmingAgent:
    def __init__(self, api_key: str = "pf_live_<your_key>"):
        self.api_key = api_key
        self.headers = {"Authorization": f"Bearer {api_key}"}

    async def get_wallet_balances(self) -> dict:
        async with httpx.AsyncClient() as c:
            r = await c.get(
                f"{PURPLE_FLEA_API}/v1/wallet/balances",
                headers=self.headers
            )
            return r.json()

    async def approve_token(self, token: str, spender: str, chain: str) -> dict:
        """Approve a DeFi contract to spend a token."""
        async with httpx.AsyncClient() as c:
            r = await c.post(
                f"{PURPLE_FLEA_API}/v1/wallet/approve",
                headers=self.headers,
                json={"token": token, "spender": spender, "chain": chain}
            )
            return r.json()

    async def deposit_to_yield_pool(
        self,
        protocol: Literal["aave", "uniswap_v3", "curve"],
        pool_id: str,
        amount_usd: float,
        chain: str = "ethereum",
    ) -> dict:
        """Deposit into a yield-bearing pool via Purple Flea's DeFi router."""
        async with httpx.AsyncClient() as c:
            r = await c.post(
                f"{PURPLE_FLEA_API}/v1/defi/deposit",
                headers=self.headers,
                json={
                    "protocol": protocol,
                    "pool_id": pool_id,
                    "amount_usd": amount_usd,
                    "chain": chain,
                    "slippage_bps": 50,  # 0.5%
                },
                timeout=30.0
            )
            return r.json()

    async def withdraw_from_pool(
        self,
        position_id: str,
        pct: float = 100.0
    ) -> dict:
        """Withdraw from an existing DeFi position."""
        async with httpx.AsyncClient() as c:
            r = await c.post(
                f"{PURPLE_FLEA_API}/v1/defi/withdraw",
                headers=self.headers,
                json={"position_id": position_id, "percentage": pct}
            )
            return r.json()

    async def harvest_rewards(self, position_id: str) -> dict:
        """Collect pending yield rewards and optionally auto-reinvest."""
        async with httpx.AsyncClient() as c:
            r = await c.post(
                f"{PURPLE_FLEA_API}/v1/defi/harvest",
                headers=self.headers,
                json={"position_id": position_id, "reinvest": True}
            )
            return r.json()

Autonomous Rebalancing Logic

The core of a yield farming agent is its rebalancing logic: when should it move capital from one pool to another? The naive answer is "always move to the highest APY pool," but this ignores gas costs, slippage, and the mean-reverting nature of APY spikes.

import asyncio
import time
from dataclasses import dataclass

@dataclass
class ActivePosition:
    position_id: str
    protocol: str
    pool_id: str
    amount_usd: float
    entry_apy: float
    entry_time: float

class YieldRebalancer:
    def __init__(
        self,
        agent: YieldFarmingAgent,
        min_apy_improvement: float = 3.0,     # need 3% APY gain to justify move
        min_hold_hours: float = 24.0,          # hold at least 24h before rebalancing
        gas_cost_usd: float = 20.0,            # estimated move cost
        max_positions: int = 5,
    ):
        self.agent = agent
        self.min_apy_improvement = min_apy_improvement
        self.min_hold_hours = min_hold_hours
        self.gas_cost_usd = gas_cost_usd
        self.max_positions = max_positions
        self.positions: list[ActivePosition] = []

    def should_rebalance(
        self,
        current_pos: ActivePosition,
        new_apy: float,
    ) -> tuple[bool, str]:
        # Must hold minimum time
        hours_held = (time.time() - current_pos.entry_time) / 3600
        if hours_held < self.min_hold_hours:
            return False, f"too early ({hours_held:.1f}h < {self.min_hold_hours}h)"

        # New APY must be meaningfully better
        improvement = new_apy - current_pos.entry_apy
        if improvement < self.min_apy_improvement:
            return False, f"APY gain insufficient ({improvement:.1f}% < {self.min_apy_improvement}%)"

        # Move must be profitable within 30 days
        daily_gain = current_pos.amount_usd * (improvement / 100) / 365
        days_to_breakeven = self.gas_cost_usd / daily_gain if daily_gain > 0 else 999
        if days_to_breakeven > 30:
            return False, f"breakeven too far ({days_to_breakeven:.0f}d > 30d)"

        return True, "rebalance approved"

    async def run_rebalance_cycle(self):
        """One iteration of the rebalancing loop."""
        pools = await fetch_defi_yields(min_tvl=10_000_000, min_apy=5.0)
        best_pool = pools[0] if pools else None
        if not best_pool:
            return

        for pos in self.positions[:]:
            ok, reason = self.should_rebalance(pos, best_pool.apy)
            if ok:
                print(f"[REBALANCE] Moving ${pos.amount_usd:.0f} from {pos.protocol} ({pos.entry_apy:.1f}% APY) to {best_pool.protocol} ({best_pool.apy:.1f}% APY)")
                await self.agent.withdraw_from_pool(pos.position_id)
                deposit_result = await self.agent.deposit_to_yield_pool(
                    protocol=best_pool.protocol,
                    pool_id=best_pool.pool_id,
                    amount_usd=pos.amount_usd,
                )
                self.positions.remove(pos)
                self.positions.append(ActivePosition(
                    position_id=deposit_result["position_id"],
                    protocol=best_pool.protocol,
                    pool_id=best_pool.pool_id,
                    amount_usd=pos.amount_usd,
                    entry_apy=best_pool.apy,
                    entry_time=time.time(),
                ))
            else:
                print(f"[HOLD] {pos.protocol}: {reason}")

Cross-Chain Yield Hunting

Some of the highest yields exist on L2s and alternative chains (Arbitrum, Optimism, Base, Solana) where liquidity is thinner and protocols compete more aggressively for TVL. Agents can bridge capital automatically using aggregators like LI.FI or Across Protocol.

async def find_best_cross_chain_yield(
    min_tvl: float = 2_000_000,
    chains: list[str] = ["Ethereum", "Arbitrum", "Optimism", "Base", "Solana"]
) -> list[dict]:
    """
    Find highest APY pools across all chains, accounting for bridge costs.
    Bridge cost is approximated as a one-time $10 gas + 0.05% fee.
    """
    all_pools = await fetch_defi_yields(min_tvl=min_tvl, chains=chains)
    BRIDGE_COST_USD = 10.0
    BRIDGE_FEE_BPS = 5  # 0.05%

    ranked = []
    for pool in all_pools[:50]:
        # Estimate time-to-breakeven bridge cost (assuming $5,000 position, 90 days)
        position_usd = 5_000
        bridge_total = BRIDGE_COST_USD + position_usd * (BRIDGE_FEE_BPS / 10_000)
        daily_yield = position_usd * (pool.apy / 100) / 365
        breakeven_days = bridge_total / daily_yield if daily_yield > 0 else 999
        ranked.append({
            "pool": pool,
            "bridge_cost_usd": bridge_total,
            "breakeven_days": breakeven_days,
            "net_90d_yield": daily_yield * 90 - bridge_total,
        })

    return sorted(ranked, key=lambda x: x["net_90d_yield"], reverse=True)

Risk Management Framework

DeFi positions carry smart contract risk, liquidation risk (for leveraged positions), and regulatory risk in addition to IL. Agents should implement hard limits.

class DeFiRiskGuard:
    PROTOCOL_WHITELIST = {
        "aave-v3", "uniswap-v3", "curve", "compound-v3",
        "gmx", "pendle", "convex-finance"
    }
    MAX_POSITION_PCT = 0.30   # 30% of portfolio per pool
    MAX_LTV = 0.60
    MIN_PROTOCOL_AGE_DAYS = 180

    def approve_deposit(
        self,
        protocol: str,
        il_risk: str,
        fee_apy: float,
        total_apy: float,
        portfolio_usd: float,
        deposit_usd: float,
    ) -> tuple[bool, str]:
        if protocol.lower() not in self.PROTOCOL_WHITELIST:
            return False, f"protocol not whitelisted: {protocol}"
        if deposit_usd / portfolio_usd > self.MAX_POSITION_PCT:
            return False, f"position too large: {deposit_usd/portfolio_usd:.0%} > {self.MAX_POSITION_PCT:.0%}"
        # For high-IL pools, require fee APY (not token rewards) to cover IL
        if il_risk == "high":
            # estimate max IL over 30d assuming 50% price swing
            est_il = abs(impermanent_loss(1.5))  # ~20% price change
            if fee_apy / 100 / 12 < est_il:  # monthly fee vs expected IL
                return False, f"fee APY too low to cover estimated IL"
        return True, "approved"

Conclusion

Yield farming is one of the clearest demonstrations of agent superiority over human capital management: a well-designed agent compounds continuously, rebalances before humans notice, and never sleeps through a high-yield spike. The combination of the DeFi Llama API for opportunity discovery, web3.py for on-chain execution, and Purple Flea's managed wallet for simplified key management gives agents a complete toolkit.

The key discipline is risk management: whitelist protocols, cap concentration, and always sell reward tokens on schedule rather than holding for speculative appreciation. The agents that survive long-term are those that optimize for risk-adjusted yield, not raw APY.