Cross-Chain Yield Strategies for AI Agents

Yield is not equal across blockchains. At any given moment, a USDC position earns 4% on Ethereum, 9% on Arbitrum, and 14% on a newer L2 with bootstrapping incentives. An AI agent capable of detecting and routing capital across these differentials has a structural advantage over static human allocators.

This guide covers how autonomous agents can implement cross-chain yield optimization: identifying opportunities, managing bridge risk, automating position rotation, and using Purple Flea's wallet infrastructure as the coordination layer. All examples are production-ready Python.

The Multi-Chain Yield Landscape

Cross-chain yield falls into four broad categories, each with distinct risk profiles and automation requirements:

Current Yield Benchmarks by Chain (March 2026)

The 9-percentage-point spread between Ethereum staking and the top Optimism farms represents real alpha — if an agent can capture it while managing the additional risks.

Building a Yield Scanner

The first component of any cross-chain yield agent is a scanner that continuously monitors available yields across protocols and chains, adjusting for gas costs and bridge fees.

pythonimport asyncio
import httpx
from dataclasses import dataclass
from typing import Optional

@dataclass
class YieldOpportunity:
    chain: str
    protocol: str
    pool: str
    base_apr: float        # Raw APR
    reward_apr: float      # Incentive token APR
    tvl_usd: float
    min_deposit_usd: float
    withdrawal_delay_days: float
    risk_score: float      # 0 = safe, 1 = high risk

    @property
    def total_apr(self) -> float:
        return self.base_apr + self.reward_apr

    @property
    def risk_adjusted_apr(self) -> float:
        """Penalize risky positions: apr * (1 - risk_score * 0.5)"""
        return self.total_apr * (1 - self.risk_score * 0.5)

    @property
    def liquidity_score(self) -> float:
        """Higher TVL = more liquidity = easier to exit."""
        return min(1.0, self.tvl_usd / 50_000_000)  # Max at $50M TVL


class CrossChainYieldScanner:
    """Aggregates yield data from multiple on-chain data providers."""

    DEFI_LLAMA_URL = "https://yields.llama.fi/pools"

    # Whitelist of trusted protocols per chain
    TRUSTED_PROTOCOLS = {
        "ethereum": ["lido", "aave-v3", "compound-v3", "uniswap-v3", "curve"],
        "polygon":  ["aave-v3", "quickswap-v3", "curve"],
        "base":     ["aerodrome", "aave-v3", "compound-v3"],
        "arbitrum": ["gmx", "aave-v3", "uniswap-v3", "radiant"],
        "solana":   ["marinade", "jito", "raydium", "orca"],
        "optimism": ["velodrome", "aave-v3", "synthetix"],
    }

    # Approximate bridge cost (USD) to reach each chain from Ethereum
    BRIDGE_COSTS = {
        "ethereum": 0,
        "polygon": 0.5,
        "base": 0.3,
        "arbitrum": 0.4,
        "solana": 2.0,
        "optimism": 0.3,
    }

    def __init__(self, min_tvl_usd: float = 5_000_000):
        self.min_tvl = min_tvl_usd

    async def fetch_all(self) -> list[YieldOpportunity]:
        async with httpx.AsyncClient(timeout=20.0) as client:
            resp = await client.get(self.DEFI_LLAMA_URL)
            resp.raise_for_status()
            pools = resp.json().get("data", [])

        opportunities = []
        for pool in pools:
            chain = pool.get("chain", "").lower()
            project = pool.get("project", "").lower()
            trusted = self.TRUSTED_PROTOCOLS.get(chain, [])
            if not any(t in project for t in trusted):
                continue
            tvl = pool.get("tvlUsd", 0)
            if tvl < self.min_tvl:
                continue

            opp = YieldOpportunity(
                chain=chain,
                protocol=project,
                pool=pool.get("symbol", ""),
                base_apr=pool.get("apyBase", 0) / 100,
                reward_apr=pool.get("apyReward", 0) / 100,
                tvl_usd=tvl,
                min_deposit_usd=pool.get("minDepositUsd", 10),
                withdrawal_delay_days=0.0,  # Enriched separately
                risk_score=self._risk_score(pool),
            )
            opportunities.append(opp)

        return sorted(opportunities, key=lambda o: o.risk_adjusted_apr, reverse=True)

    def _risk_score(self, pool: dict) -> float:
        """Heuristic risk score based on pool metadata."""
        score = 0.0
        # High reward APR suggests unsustainable token incentives
        if pool.get("apyReward", 0) > 50:
            score += 0.3
        # Low TVL = harder to exit, more manipulation risk
        if pool.get("tvlUsd", 0) < 10_000_000:
            score += 0.2
        # Newer pools have less audit history
        if not pool.get("audits"):
            score += 0.2
        # IL risk for non-stablecoin pairs
        if "USDC" not in pool.get("symbol", "") and "USDT" not in pool.get("symbol", ""):
            score += 0.15
        return min(score, 1.0)

    def net_apr_after_bridge(self, opp: YieldOpportunity, capital: float,
                             hold_days: float = 30) -> float:
        """APR net of bridge cost for a given capital amount and hold period."""
        bridge_cost = self.BRIDGE_COSTS.get(opp.chain, 1.0)
        bridge_drag = bridge_cost / capital  # As fraction of capital
        daily_gross = opp.total_apr / 365
        net_daily = daily_gross - bridge_drag / hold_days
        return net_daily * 365

Bridge Strategy and Risk Management

Bridges are the highest-risk component of cross-chain yield. Over $2 billion has been lost to bridge exploits. Agents must apply strict bridge selection criteria and diversify across bridges when capital warrants it.

pythonfrom enum import Enum
from dataclasses import dataclass

class BridgeTier(Enum):
    TIER_1 = "tier_1"   # Battle-tested, audited, $1B+ TVL
    TIER_2 = "tier_2"   # Audited, $100M+ TVL, 12+ months live
    TIER_3 = "tier_3"   # Newer, audited, $10M+ TVL

@dataclass
class Bridge:
    name: str
    supported_chains: list[str]
    fee_bps: int          # Fee in basis points
    avg_time_seconds: int
    tier: BridgeTier
    max_agent_allocation: float  # Max fraction of portfolio to bridge at once

APPROVED_BRIDGES = [
    Bridge("Stargate",     ["ethereum","polygon","base","arbitrum","optimism"], 6,  60, BridgeTier.TIER_1, 0.20),
    Bridge("Across",       ["ethereum","polygon","base","arbitrum","optimism"], 5,  120, BridgeTier.TIER_1, 0.20),
    Bridge("Hop Protocol", ["ethereum","polygon","base","arbitrum","optimism"], 4,  300, BridgeTier.TIER_2, 0.15),
    Bridge("Wormhole",     ["ethereum","solana","polygon","base"],              10, 900, BridgeTier.TIER_1, 0.15),
    Bridge("deBridge",     ["ethereum","solana","arbitrum"],                   8,  180, BridgeTier.TIER_2, 0.10),
]

def select_bridge(source: str, destination: str, amount_usd: float,
                  portfolio_total: float) -> Bridge | None:
    """Select the optimal bridge for a given route and amount."""
    eligible = [
        b for b in APPROVED_BRIDGES
        if source in b.supported_chains and destination in b.supported_chains
        and amount_usd / portfolio_total <= b.max_agent_allocation
    ]
    if not eligible:
        return None  # No safe bridge found — do not bridge

    # Prefer Tier 1, then lowest fee
    tier1 = [b for b in eligible if b.tier == BridgeTier.TIER_1]
    pool = tier1 if tier1 else eligible
    return min(pool, key=lambda b: b.fee_bps)

def bridge_fee_usd(bridge: Bridge, amount_usd: float) -> float:
    return amount_usd * bridge.fee_bps / 10_000

def min_hold_days_to_break_even(bridge: Bridge, amount_usd: float,
                                 net_daily_alpha: float) -> float:
    """How many days before bridging is worth it over staying on source chain."""
    fee = bridge_fee_usd(bridge, amount_usd)
    if net_daily_alpha <= 0:
        return float("inf")  # Never worth bridging
    return fee / (amount_usd * net_daily_alpha / 365)

# Example: Is it worth bridging $10,000 to Base for 11.3% APR vs Ethereum 4.2%?
bridge = select_bridge("ethereum", "base", 10_000, 50_000)
if bridge:
    daily_alpha = (0.113 - 0.042) / 365  # 7.1% annual alpha
    breakeven = min_hold_days_to_break_even(bridge, 10_000, daily_alpha)
    print(f"Bridge: {bridge.name}, Fee: ${bridge_fee_usd(bridge, 10_000):.2f}")
    print(f"Break-even after: {breakeven:.1f} days")
    # If planning to hold 30+ days, bridging is clearly worth it.

Automated Position Rotation

The core loop of a cross-chain yield agent: scan opportunities, compare with current positions, and rotate capital when the net benefit exceeds a rotation threshold.

pythonimport asyncio
import httpx
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class YieldPosition:
    chain: str
    protocol: str
    pool: str
    capital_usd: float
    entry_apr: float
    entry_timestamp: float
    days_held: float = 0.0

    def current_daily_earnings(self) -> float:
        return self.capital_usd * self.entry_apr / 365

class CrossChainYieldAgent:
    """
    Autonomous agent that rotates capital across chains to maximize yield.
    Uses Purple Flea wallet API for signing and execution.
    """

    ROTATION_THRESHOLD = 0.03  # Minimum APR improvement (3%) to justify rotation
    MIN_HOLD_DAYS = 7           # Never rotate before 7 days (avoid churn)
    MAX_POSITIONS = 5           # Diversify across at most 5 positions
    MIN_POSITION_USD = 100      # Minimum position size

    def __init__(self, api_key: str, total_capital_usd: float):
        self.api_key = api_key  # pf_live_
        self.total_capital = total_capital_usd
        self.positions: list[YieldPosition] = []
        self.scanner = CrossChainYieldScanner()

    async def run_cycle(self):
        """Execute one yield optimization cycle."""
        print(f"\n[{asyncio.get_event_loop().time():.0f}] Starting yield cycle...")

        # 1. Scan current opportunities
        opportunities = await self.scanner.fetch_all()
        if not opportunities:
            print("No opportunities found, skipping cycle")
            return

        top_opp = opportunities[0]
        print(f"Best opportunity: {top_opp.chain}/{top_opp.protocol}/{top_opp.pool} "
              f"@ {top_opp.risk_adjusted_apr:.1%} risk-adj APR")

        # 2. Evaluate each current position for rotation
        rotations = []
        for pos in self.positions:
            if pos.days_held < self.MIN_HOLD_DAYS:
                continue  # Too early to rotate

            net_improvement = top_opp.risk_adjusted_apr - pos.entry_apr
            bridge = select_bridge(pos.chain, top_opp.chain, pos.capital_usd,
                                   self.total_capital)
            if bridge is None:
                continue

            breakeven = min_hold_days_to_break_even(
                bridge, pos.capital_usd,
                net_improvement if net_improvement > 0 else 0
            )
            # Only rotate if we expect to hold long enough to break even
            if net_improvement > self.ROTATION_THRESHOLD and breakeven < 14:
                rotations.append((pos, top_opp, bridge))

        # 3. Execute rotations
        for pos, target, bridge in rotations:
            await self._rotate(pos, target, bridge)

        # 4. Deploy any idle capital
        idle = self._idle_capital()
        if idle > self.MIN_POSITION_USD and len(self.positions) < self.MAX_POSITIONS:
            await self._deploy_new_position(top_opp, idle)

    async def _rotate(self, pos: YieldPosition, target: YieldOpportunity, bridge: Bridge):
        """Withdraw from current position and bridge to new chain/protocol."""
        print(f"Rotating: {pos.chain}/{pos.protocol} -> {target.chain}/{target.protocol}")
        # 1. Withdraw from current protocol
        await self._wallet_call("withdraw", {
            "chain": pos.chain,
            "protocol": pos.protocol,
            "pool": pos.pool,
            "amount_usd": pos.capital_usd,
        })
        # 2. Bridge to target chain
        await self._wallet_call("bridge", {
            "bridge": bridge.name,
            "source_chain": pos.chain,
            "dest_chain": target.chain,
            "amount_usd": pos.capital_usd,
        })
        # 3. Deposit into new protocol
        await self._wallet_call("deposit", {
            "chain": target.chain,
            "protocol": target.protocol,
            "pool": target.pool,
            "amount_usd": pos.capital_usd,
        })
        # Update position record
        pos.chain = target.chain
        pos.protocol = target.protocol
        pos.pool = target.pool
        pos.entry_apr = target.total_apr
        pos.days_held = 0.0

    async def _wallet_call(self, action: str, params: dict):
        """Execute a wallet action via Purple Flea API."""
        async with httpx.AsyncClient(timeout=30.0) as client:
            r = await client.post(
                f"https://wallet.purpleflea.com/api/{action}",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json=params
            )
            r.raise_for_status()
            return r.json()

    def _idle_capital(self) -> float:
        deployed = sum(p.capital_usd for p in self.positions)
        # Keep 20% as operating reserve
        max_deployable = self.total_capital * 0.80
        return max(0, max_deployable - deployed)

    async def _deploy_new_position(self, opp: YieldOpportunity, amount: float):
        print(f"Deploying ${amount:.2f} into {opp.chain}/{opp.protocol}/{opp.pool}")
        await self._wallet_call("deposit", {
            "chain": opp.chain, "protocol": opp.protocol,
            "pool": opp.pool, "amount_usd": amount,
        })
        import time
        self.positions.append(YieldPosition(
            chain=opp.chain, protocol=opp.protocol, pool=opp.pool,
            capital_usd=amount, entry_apr=opp.total_apr,
            entry_timestamp=time.time()
        ))

Impermanent Loss Management

DEX liquidity provision earns trading fees but exposes agents to impermanent loss (IL) — the opportunity cost of holding the LP position versus holding the underlying tokens. Agents must quantify IL continuously and factor it into rotation decisions.

pythonimport math

def impermanent_loss(price_ratio_change: float) -> float:
    """
    Calculate impermanent loss as a fraction of held position value.
    price_ratio_change: ratio of final price to initial price (e.g., 2.0 for 2x)
    Returns: IL as a decimal (e.g., 0.057 = 5.7% loss)
    """
    k = math.sqrt(price_ratio_change)
    il = 2 * k / (1 + price_ratio_change) - 1
    return abs(il)

def lp_net_return(
    fee_apr: float,
    reward_apr: float,
    avg_price_volatility_daily: float,
    days: int
) -> float:
    """
    Estimate net return for an LP position, accounting for expected IL.
    avg_price_volatility_daily: typical daily price move (e.g., 0.02 for 2%)
    """
    gross_apr = fee_apr + reward_apr
    # Expected IL scales approximately with variance
    daily_variance = avg_price_volatility_daily ** 2
    annual_il = daily_variance * 365 * 0.5  # Approximation
    net_apr = gross_apr - annual_il
    return net_apr * days / 365

# Example: USDC/ETH pool on Base
# ETH moves 3% daily on average, fee APR = 15%, reward APR = 8%
net = lp_net_return(fee_apr=0.15, reward_apr=0.08, avg_price_volatility_daily=0.03, days=30)
print(f"Expected 30-day return: {net:.2%}")  # Net of expected IL

# IL lookup table for common scenarios
scenarios = [(1.25, "25% price move"), (2.0, "2x price"), (4.0, "4x price"), (0.5, "50% drop")]
for ratio, label in scenarios:
    print(f"  IL at {label}: {impermanent_loss(ratio):.2%}")

# Recommendation: for volatile pairs, only LP if fee_apr + reward_apr > 2x expected IL
def should_lp(fee_apr: float, reward_apr: float, daily_vol: float) -> bool:
    expected_annual_il = daily_vol ** 2 * 365 * 0.5
    return (fee_apr + reward_apr) > 2 * expected_annual_il

Purple Flea Wallet as Yield Coordination Layer

Purple Flea's wallet API provides multi-chain key management that makes cross-chain yield automation practical. Instead of managing individual private keys per chain, the agent uses a single API key to sign and broadcast transactions across all supported chains.

pythonimport httpx
import asyncio

class PurpleFlealYieldWallet:
    """
    Wrapper around Purple Flea wallet API for yield strategy execution.
    Supports: Ethereum, Polygon, Base, Arbitrum, Optimism, Solana.
    """

    BASE_URL = "https://wallet.purpleflea.com/api"

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

    async def get_balances(self, chains: list[str] = None) -> dict:
        """Get USDC balances across all chains."""
        async with httpx.AsyncClient(timeout=15.0) as client:
            r = await client.get(
                f"{self.BASE_URL}/balances",
                headers=self.headers,
                params={"chains": ",".join(chains or ["ethereum","polygon","base","arbitrum"])}
            )
            r.raise_for_status()
            return r.json()["balances"]  # {"ethereum": 500.0, "base": 250.0, ...}

    async def bridge_usdc(self, source: str, destination: str,
                           amount_usd: float, bridge_name: str) -> dict:
        """Bridge USDC between chains. Returns tx hash."""
        async with httpx.AsyncClient(timeout=60.0) as client:
            r = await client.post(
                f"{self.BASE_URL}/bridge",
                headers=self.headers,
                json={"source_chain": source, "dest_chain": destination,
                      "amount_usdc": amount_usd, "bridge": bridge_name}
            )
            r.raise_for_status()
            return r.json()  # {"tx_hash": "0x...", "estimated_arrival_seconds": 120}

    async def deposit_to_protocol(self, chain: str, protocol: str,
                                   pool: str, amount_usd: float) -> dict:
        """Deposit USDC into a yield protocol. Returns position ID."""
        async with httpx.AsyncClient(timeout=30.0) as client:
            r = await client.post(
                f"{self.BASE_URL}/defi/deposit",
                headers=self.headers,
                json={"chain": chain, "protocol": protocol,
                      "pool": pool, "amount_usdc": amount_usd}
            )
            r.raise_for_status()
            return r.json()

    async def withdraw_from_protocol(self, chain: str, position_id: str,
                                      amount_usd: float = None) -> dict:
        """Withdraw from a yield position. Pass None for full withdrawal."""
        async with httpx.AsyncClient(timeout=30.0) as client:
            r = await client.post(
                f"{self.BASE_URL}/defi/withdraw",
                headers=self.headers,
                json={"chain": chain, "position_id": position_id,
                      "amount_usdc": amount_usd}  # None = full withdrawal
            )
            r.raise_for_status()
            return r.json()

    async def claim_rewards(self, chain: str, position_id: str) -> dict:
        """Claim accumulated yield/reward tokens from a position."""
        async with httpx.AsyncClient(timeout=20.0) as client:
            r = await client.post(
                f"{self.BASE_URL}/defi/claim",
                headers=self.headers,
                json={"chain": chain, "position_id": position_id}
            )
            r.raise_for_status()
            return r.json()  # {"claimed_usdc": 12.50, "claimed_tokens": {"OP": 5.2}}

Cross-Chain Risk Framework

Cross-chain operations introduce risks beyond standard DeFi: bridge exploits, chain reorganizations, gas token unavailability, and protocol-specific risks multiplied across chains. A robust agent applies a risk budget that limits total exposure to any single failure mode.

pythonclass CrossChainRiskManager:
    """Enforce cross-chain risk limits before any capital movement."""

    CHAIN_LIMIT     = 0.50   # Max 50% on any single chain
    PROTOCOL_LIMIT  = 0.30   # Max 30% in any single protocol
    BRIDGE_TX_LIMIT = 0.20   # Max 20% per bridge transaction
    GAS_BUFFER_USD  = 5.0    # Min gas reserve per active chain

    def __init__(self, total_capital: float):
        self.total = total_capital
        self.positions: dict = {}    # {chain: {protocol: amount}}
        self.gas_reserves: dict = {} # {chain: amount}

    def can_add_position(self, chain: str, protocol: str, amount: float) -> tuple[bool, str]:
        """Check if adding a position would violate risk limits."""
        chain_total = sum(self.positions.get(chain, {}).values()) + amount
        if chain_total / self.total > self.CHAIN_LIMIT:
            return False, f"Chain limit: would reach {chain_total/self.total:.0%} on {chain}"

        protocol_total = self.positions.get(chain, {}).get(protocol, 0) + amount
        if protocol_total / self.total > self.PROTOCOL_LIMIT:
            return False, f"Protocol limit: would reach {protocol_total/self.total:.0%} in {protocol}"

        gas = self.gas_reserves.get(chain, 0)
        if gas < self.GAS_BUFFER_USD:
            return False, f"Insufficient gas on {chain}: ${gas:.2f} (need ${self.GAS_BUFFER_USD})"

        return True, "OK"

    def can_bridge(self, source: str, dest: str, amount: float) -> tuple[bool, str]:
        if amount / self.total > self.BRIDGE_TX_LIMIT:
            return False, f"Bridge tx limit: {amount/self.total:.0%} exceeds {self.BRIDGE_TX_LIMIT:.0%}"
        gas = self.gas_reserves.get(source, 0)
        if gas < self.GAS_BUFFER_USD:
            return False, f"Insufficient gas on source chain {source}"
        return True, "OK"

    def diversification_score(self) -> float:
        """0 = fully concentrated, 1 = perfectly diversified across 4+ chains."""
        if not self.positions:
            return 0.0
        chain_fractions = []
        for chain, protocols in self.positions.items():
            chain_fractions.append(sum(protocols.values()) / self.total)
        hhi = sum(f ** 2 for f in chain_fractions)  # Herfindahl index
        return 1 - hhi  # Higher = more diversified

Automated Reward Harvesting

Many yield protocols distribute rewards in their native governance tokens (OP, ARB, COMP, etc.). These tokens depreciate over time as new emissions dilute holders. Agents should auto-harvest and swap rewards to USDC on a regular schedule.

pythonimport asyncio

class RewardHarvester:
    """
    Automatically claims and converts governance token rewards to USDC.
    Runs on a configurable schedule (default: every 24 hours).
    """

    HARVEST_INTERVAL_HOURS = 24
    MIN_HARVEST_USD = 1.0    # Skip harvest if rewards worth less than $1

    def __init__(self, wallet: PurpleFlealYieldWallet):
        self.wallet = wallet
        self.positions: list[dict] = []  # [{chain, position_id}]

    async def harvest_all(self) -> float:
        """Claim all pending rewards and convert to USDC. Returns total USDC earned."""
        total_usdc = 0.0
        for pos in self.positions:
            try:
                result = await self.wallet.claim_rewards(pos["chain"], pos["position_id"])
                claimed_usdc = result.get("claimed_usdc", 0)
                claimed_tokens = result.get("claimed_tokens", {})

                # Convert any governance tokens to USDC immediately
                for token, amount in claimed_tokens.items():
                    if amount > 0:
                        usdc_value = await self._swap_to_usdc(pos["chain"], token, amount)
                        claimed_usdc += usdc_value

                if claimed_usdc >= self.MIN_HARVEST_USD:
                    total_usdc += claimed_usdc
                    print(f"Harvested ${claimed_usdc:.2f} USDC from {pos['chain']}/{pos['position_id']}")

            except Exception as e:
                print(f"Harvest failed for {pos}: {e}")

        return total_usdc

    async def _swap_to_usdc(self, chain: str, token: str, amount: float) -> float:
        """Swap a reward token to USDC via Purple Flea trading API."""
        async with httpx.AsyncClient(timeout=20.0) as client:
            r = await client.post(
                "https://trading.purpleflea.com/api/swap",
                headers=self.wallet.headers,
                json={"chain": chain, "from_token": token,
                      "to_token": "USDC", "amount": amount, "slippage_bps": 50}
            )
            if r.status_code == 200:
                return r.json().get("received_usdc", 0)
            return 0.0

    async def run_forever(self):
        """Background harvesting loop."""
        while True:
            total = await self.harvest_all()
            print(f"Harvest cycle complete: ${total:.2f} USDC earned")
            await asyncio.sleep(self.HARVEST_INTERVAL_HOURS * 3600)

Conclusion

Cross-chain yield optimization is one of the highest-value strategies available to autonomous AI agents. The key advantages are time and scale: agents can monitor dozens of chains continuously, react to yield shifts within minutes rather than days, and execute complex multi-step rotations without human intervention.

The framework in this guide — yield scanning, bridge selection, IL-aware LP evaluation, rotation logic, and automated reward harvesting — gives agents a complete playbook. The Purple Flea wallet API provides the execution layer, abstracting away the complexity of managing keys and signing transactions across every chain.

Start with a single chain and one or two trusted protocols. Add cross-chain exposure gradually as you build confidence in your risk management implementation. The alpha is real — but only for agents disciplined enough to respect the risk limits.