Cross-Chain Agent Strategies: Bridging, Arbitrage, and Yield Hunting Across Networks

Why Cross-Chain Is Now Table Stakes for Agents

The multi-chain ecosystem has matured dramatically. Ethereum mainnet, Solana, Arbitrum, and Base collectively process billions in daily volume — and prices, yields, and liquidity frequently diverge across them. An agent confined to a single chain is operating with one hand tied behind its back.

Consider a simple observation: USDC yield on Arbitrum's top lending protocol regularly runs 1–3% higher than on Ethereum mainnet. Over a year, that gap compounds into meaningful alpha. Or consider that token prices on low-latency chains like Solana can lag Ethereum price discovery by 30–120 seconds — a window an autonomous agent can reliably exploit.

This guide builds a complete CrossChainAgent in Python that:

• Monitors prices and yields across four chains simultaneously
• Selects the optimal bridge for each transfer (cost, speed, reliability)
• Executes cross-chain arbitrage when spreads exceed bridge costs
• Rebalances capital weekly to chase the highest on-chain yields
• Tracks all multi-chain balances through the Purple Flea Wallet API

The Four-Chain Landscape

Each chain in a cross-chain agent's universe has distinct characteristics that determine when and why capital should live there:

Bridge Protocols: Cost, Speed, and Reliability

The bridge selection is the most critical decision in any cross-chain strategy. Bridge fees directly eat into arbitrage spreads, and bridge delays affect time-sensitivity of trades. Agents must select bridges dynamically based on trade urgency and size.

Cross-Chain Arbitrage Mechanics

Cross-chain arbitrage exploits price differences for the same asset on different chains. Unlike on-chain DEX arbitrage (which can be completed atomically in one block), cross-chain arb involves a 15–300 second bridge latency — meaning the agent takes price risk during transit.

Price Gap Detection

The agent must scan prices continuously and calculate whether the gap exceeds the total cost of the trade:

Transit Price Risk

The most dangerous aspect of cross-chain arb is that the opportunity may close during the bridge transit. Strategies to mitigate this:

• Minimum spread filter — only execute when gap is 3x the typical spread volatility over the bridge duration
• Use fastest bridges — Across Protocol's 15-second finality dramatically reduces transit risk
• Partial hedging — short the sell-side asset on a perpetuals exchange while the bridge transfer completes
• Historical spread analysis — calculate the median spread reversion time per asset pair and only enter if spread is statistically likely to persist

Full Python CrossChainAgent Implementation

import asyncio
import time
import json
import httpx
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

# ── Constants ──────────────────────────────────────────────────────────────
API_BASE     = "https://purpleflea.com/api"
WALLET_API   = "https://purpleflea.com/api/wallet"
TRADING_API  = "https://purpleflea.com/api/trading"
API_KEY      = "pf_live_your_key_here"   # Register at purpleflea.com/for-agents

class Chain(Enum):
    ETHEREUM = "ethereum"
    SOLANA   = "solana"
    ARBITRUM = "arbitrum"
    BASE     = "base"

class Bridge(Enum):
    STARGATE = "stargate"
    ACROSS   = "across"
    WORMHOLE = "wormhole"
    CCTP     = "cctp"           # Circle native USDC

# Bridge fee table (approximate pct + fixed USD gas)
BRIDGE_FEES: dict[Bridge, dict] = {
    Bridge.STARGATE: {"pct": 0.0006, "fixed_usd": 2.0,  "avg_seconds": 60},
    Bridge.ACROSS:   {"pct": 0.0005, "fixed_usd": 1.0,  "avg_seconds": 20},
    Bridge.WORMHOLE: {"pct": 0.0010, "fixed_usd": 3.0,  "avg_seconds": 120},
    Bridge.CCTP:     {"pct": 0.0000, "fixed_usd": 0.5,  "avg_seconds": 180},
}

# Which bridge works for which chain pairs
BRIDGE_SUPPORT: dict[tuple[Chain, Chain], list[Bridge]] = {
    (Chain.ETHEREUM, Chain.ARBITRUM): [Bridge.ACROSS, Bridge.STARGATE, Bridge.CCTP],
    (Chain.ETHEREUM, Chain.BASE):     [Bridge.ACROSS, Bridge.STARGATE, Bridge.CCTP],
    (Chain.ARBITRUM, Chain.BASE):     [Bridge.ACROSS, Bridge.STARGATE],
    (Chain.ETHEREUM, Chain.SOLANA):   [Bridge.WORMHOLE],
    (Chain.ARBITRUM, Chain.SOLANA):   [Bridge.WORMHOLE],
    (Chain.BASE, Chain.SOLANA):       [Bridge.WORMHOLE],
}


@dataclass
class ChainState:
    chain: Chain
    prices: dict[str, float]        = field(default_factory=dict)
    yields: dict[str, float]        = field(default_factory=dict)   # protocol → APY
    balances: dict[str, float]      = field(default_factory=dict)   # token → amount
    last_updated: float             = 0.0


@dataclass
class BridgeTransfer:
    tx_id:        str
    source_chain: Chain
    dest_chain:   Chain
    token:        str
    amount:       float
    bridge:       Bridge
    initiated_at: float
    expected_completion: float
    status:       str = "pending"   # pending / complete / failed


@dataclass
class ArbitrageOpportunity:
    token:      str
    buy_chain:  Chain
    sell_chain: Chain
    buy_price:  float
    sell_price: float
    gap_pct:    float
    net_profit_pct: float
    bridge:     Bridge
    trade_size: float


class CrossChainAgent:
    """
    Autonomous cross-chain agent for Purple Flea.
    Monitors ETH, SOL, ARB, Base — bridges, arbitrages, and yield-hunts.
    """

    SCAN_INTERVAL      = 30           # seconds between price scans
    YIELD_REBAL_SECS   = 7 * 86_400  # rebalance yield positions weekly
    MIN_ARB_PROFIT_PCT = 0.001        # 0.1% minimum net profit to execute arb
    MAX_TRADE_PCT      = 0.25         # max 25% of chain balance per trade
    MIN_TRADE_USD      = 500          # minimum trade size to cover fixed bridge fees
    TARGET_CHAINS      = [Chain.ETHEREUM, Chain.SOLANA, Chain.ARBITRUM, Chain.BASE]

    def __init__(self):
        self.chain_states: dict[Chain, ChainState] = {
            c: ChainState(chain=c) for c in self.TARGET_CHAINS
        }
        self.pending_bridges: list[BridgeTransfer] = []
        self.completed_arbs:  list[dict]            = []
        self.total_arb_pnl    = 0.0
        self.total_yield_pnl  = 0.0
        self.last_yield_rebal = 0.0
        self.client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {API_KEY}",
                     "Content-Type": "application/json"},
            timeout=30
        )

    # ── Data Fetching ──────────────────────────────────────────────────────

    async def scan_chains(self) -> None:
        """Fetch prices, yields, and balances across all chains in parallel."""
        tasks = [self._fetch_chain_data(chain) for chain in self.TARGET_CHAINS]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for chain, result in zip(self.TARGET_CHAINS, results):
            if isinstance(result, Exception):
                print(f"[WARN] Failed to fetch {chain.value}: {result}")
            else:
                self.chain_states[chain] = result
                print(
                    f"[SCAN] {chain.value}: "
                    f"BTC={result.prices.get('BTC', 0):.0f} "
                    f"ETH={result.prices.get('ETH', 0):.0f} "
                    f"best_yield={max(result.yields.values(), default=0):.2f}%"
                )

    async def _fetch_chain_data(self, chain: Chain) -> ChainState:
        """Fetch all data for a single chain from Purple Flea price feed."""
        state = ChainState(chain=chain)

        # Prices
        price_r = await self.client.get(
            f"{API_BASE}/prices",
            params={"chain": chain.value, "symbols": "BTC,ETH,SOL,USDC,ARB"}
        )
        price_r.raise_for_status()
        price_data = price_r.json()
        state.prices = {item["symbol"]: float(item["price"]) for item in price_data["prices"]}

        # DeFi yields on this chain
        yield_r = await self.client.get(
            f"{API_BASE}/defi/yields",
            params={"chain": chain.value, "token": "USDC"}
        )
        yield_r.raise_for_status()
        yield_data = yield_r.json()
        state.yields = {item["protocol"]: float(item["apy"]) for item in yield_data["pools"]}

        # Our balances on this chain
        bal_r = await self.client.get(
            f"{WALLET_API}/balances",
            params={"chain": chain.value}
        )
        bal_r.raise_for_status()
        bal_data = bal_r.json()
        state.balances = {item["token"]: float(item["amount"]) for item in bal_data["balances"]}

        state.last_updated = time.time()
        return state

    # ── Bridge Selection ───────────────────────────────────────────────────

    def find_bridge(
        self,
        source: Chain,
        dest: Chain,
        amount_usd: float,
        urgency: str = "normal"   # "fast" | "normal" | "cheap"
    ) -> Optional[Bridge]:
        """
        Select the optimal bridge given source/dest chain and trade urgency.
        Returns None if no bridge supports this chain pair.
        """
        pair = (source, dest)
        rev_pair = (dest, source)
        supported = BRIDGE_SUPPORT.get(pair) or BRIDGE_SUPPORT.get(rev_pair)
        if not supported:
            return None

        def bridge_score(bridge: Bridge) -> float:
            info = BRIDGE_FEES[bridge]
            total_cost = info["pct"] * amount_usd + info["fixed_usd"]
            cost_pct   = total_cost / amount_usd
            speed_score = 1.0 / info["avg_seconds"]  # higher = faster
            if urgency == "fast":
                return speed_score * 10 - cost_pct
            elif urgency == "cheap":
                return -cost_pct
            else:
                return speed_score * 3 - cost_pct

        return max(supported, key=bridge_score)

    def calculate_bridge_cost(
        self, bridge: Bridge, amount_usd: float
    ) -> tuple[float, float]:
        """Returns (total_cost_usd, cost_pct)."""
        info = BRIDGE_FEES[bridge]
        cost = info["pct"] * amount_usd + info["fixed_usd"]
        return cost, cost / amount_usd

    # ── Arbitrage Engine ───────────────────────────────────────────────────

    async def find_arbitrage_opportunities(self) -> list[ArbitrageOpportunity]:
        """Scan all chain pairs for cross-chain price gaps."""
        opportunities = []
        tokens = ["ETH", "BTC", "SOL"]

        for token in tokens:
            chain_prices = []
            for chain in self.TARGET_CHAINS:
                price = self.chain_states[chain].prices.get(token)
                if price and price > 0:
                    chain_prices.append((chain, price))

            # Check every buy/sell pair
            for i, (buy_chain, buy_price) in enumerate(chain_prices):
                for sell_chain, sell_price in chain_prices[i+1:]:
                    if sell_price <= buy_price:
                        buy_chain, sell_chain = sell_chain, buy_chain
                        buy_price, sell_price = sell_price, buy_price

                    gap_pct = (sell_price - buy_price) / buy_price

                    # Check if a bridge exists
                    bridge = self.find_bridge(buy_chain, sell_chain,
                                              10_000, urgency="fast")
                    if not bridge:
                        continue

                    # Calculate all costs
                    buy_bal_usd = (
                        self.chain_states[buy_chain].balances.get("USDC", 0)
                    )
                    trade_size  = min(
                        buy_bal_usd * self.MAX_TRADE_PCT,
                        10_000  # cap at $10K per arb
                    )
                    if trade_size < self.MIN_TRADE_USD:
                        continue

                    _, bridge_cost_pct = self.calculate_bridge_cost(bridge, trade_size)
                    dex_fees    = 0.0005 + 0.0005   # 0.05% buy + 0.05% sell
                    slippage    = trade_size / 2_000_000   # rough: $1M pool depth
                    net_pct     = gap_pct - bridge_cost_pct - dex_fees - slippage

                    if net_pct >= self.MIN_ARB_PROFIT_PCT:
                        opportunities.append(ArbitrageOpportunity(
                            token           = token,
                            buy_chain       = buy_chain,
                            sell_chain      = sell_chain,
                            buy_price       = buy_price,
                            sell_price      = sell_price,
                            gap_pct         = gap_pct,
                            net_profit_pct  = net_pct,
                            bridge          = bridge,
                            trade_size      = trade_size
                        ))

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

    async def execute_bridge(
        self,
        source: Chain,
        dest: Chain,
        token: str,
        amount: float,
        bridge: Bridge,
        reason: str = "arbitrage"
    ) -> Optional[BridgeTransfer]:
        """Initiate a cross-chain bridge transfer."""
        payload = {
            "source_chain": source.value,
            "dest_chain":   dest.value,
            "token":        token,
            "amount":       amount,
            "bridge":       bridge.value,
            "reason":       reason
        }
        try:
            r = await self.client.post(f"{API_BASE}/bridge/initiate", json=payload)
            r.raise_for_status()
            data = r.json()
            transfer = BridgeTransfer(
                tx_id           = data["tx_id"],
                source_chain    = source,
                dest_chain      = dest,
                token           = token,
                amount          = amount,
                bridge          = bridge,
                initiated_at    = time.time(),
                expected_completion = time.time() + BRIDGE_FEES[bridge]["avg_seconds"]
            )
            self.pending_bridges.append(transfer)
            print(
                f"[BRIDGE] {amount:.4f} {token} {source.value}→{dest.value} "
                f"via {bridge.value} (tx: {data['tx_id'][:12]}...)"
            )
            return transfer
        except Exception as e:
            print(f"[ERROR] Bridge failed: {e}")
            return None

    async def execute_swap(
        self,
        chain: Chain,
        from_token: str,
        to_token: str,
        amount: float,
        side: str = "buy"
    ) -> float:
        """Execute a DEX swap on the specified chain. Returns fill price."""
        payload = {
            "chain":      chain.value,
            "from_token": from_token,
            "to_token":   to_token,
            "amount":     amount,
            "slippage":   0.005   # 0.5% max slippage
        }
        r = await self.client.post(f"{TRADING_API}/swap", json=payload)
        r.raise_for_status()
        data = r.json()
        return float(data["fill_price"])

    async def calculate_bridge_pnl(self, transfer: BridgeTransfer) -> float:
        """
        After bridge completion, compute the actual profit from the arb.
        Compares sell execution price to buy price minus all fees.
        """
        if transfer.status != "complete":
            return 0.0

        # Re-check current sell-side price
        sell_state = self.chain_states[transfer.dest_chain]
        sell_price = sell_state.prices.get(transfer.token, 0)
        buy_record = next(
            (a for a in self.completed_arbs if a.get("tx_id") == transfer.tx_id), None
        )
        if not buy_record:
            return 0.0

        buy_price   = buy_record["buy_price"]
        bridge_cost, _ = self.calculate_bridge_cost(transfer.bridge, transfer.amount * buy_price)
        gross_pnl   = (sell_price - buy_price) * transfer.amount
        net_pnl     = gross_pnl - bridge_cost
        self.total_arb_pnl += net_pnl
        await self.log_income(net_pnl, "cross_chain_arb", transfer.token)
        return net_pnl

    async def run_arbitrage(self, opp: ArbitrageOpportunity):
        """Execute a full cross-chain arbitrage: buy → bridge → sell."""
        print(
            f"[ARB] {opp.token} {opp.buy_chain.value}→{opp.sell_chain.value} "
            f"gap={opp.gap_pct*100:.3f}% net={opp.net_profit_pct*100:.3f}% "
            f"size=${opp.trade_size:.0f}"
        )

        # Step 1: Buy on source chain
        fill_price = await self.execute_swap(
            opp.buy_chain, "USDC", opp.token,
            opp.trade_size / opp.buy_price
        )
        token_amount = opp.trade_size / fill_price

        # Step 2: Bridge to destination
        transfer = await self.execute_bridge(
            opp.buy_chain, opp.sell_chain,
            opp.token, token_amount, opp.bridge
        )
        if not transfer:
            return

        # Record for later PnL calculation
        self.completed_arbs.append({
            "tx_id":     transfer.tx_id,
            "buy_price": fill_price,
            "token":     opp.token,
            "size":      token_amount
        })

        # Step 3: Wait for bridge then sell (handled in monitor loop)
        print(
            f"[ARB] Bridge initiated. Expected completion in "
            f"{BRIDGE_FEES[opp.bridge]['avg_seconds']}s. "
            f"Monitoring for sell trigger..."
        )

    async def monitor_pending_bridges(self):
        """Check status of pending bridge transfers and execute sells."""
        for transfer in list(self.pending_bridges):
            if time.time() < transfer.expected_completion:
                continue

            # Check status
            try:
                r = await self.client.get(
                    f"{API_BASE}/bridge/status/{transfer.tx_id}"
                )
                r.raise_for_status()
                data = r.json()
                transfer.status = data["status"]
            except Exception as e:
                print(f"[WARN] Bridge status check failed: {e}")
                continue

            if transfer.status == "complete":
                # Sell on destination chain
                sell_price = await self.execute_swap(
                    transfer.dest_chain, transfer.token, "USDC", transfer.amount
                )
                pnl = await self.calculate_bridge_pnl(transfer)
                print(
                    f"[ARB COMPLETE] {transfer.token} sold @ {sell_price:.2f}. "
                    f"PnL: ${pnl:.2f}"
                )
                self.pending_bridges.remove(transfer)

            elif transfer.status == "failed":
                print(f"[FAIL] Bridge {transfer.tx_id[:12]}... failed. Refund pending.")
                self.pending_bridges.remove(transfer)

    # ── Yield Hunting ──────────────────────────────────────────────────────

    async def find_best_yield(self) -> tuple[Chain, str, float]:
        """
        Find the chain+protocol offering the highest USDC yield right now.
        Returns (chain, protocol, apy_pct).
        """
        best_chain    = Chain.ETHEREUM
        best_protocol = "aave-v3"
        best_apy      = 0.0

        for chain in self.TARGET_CHAINS:
            yields = self.chain_states[chain].yields
            for protocol, apy in yields.items():
                if apy > best_apy:
                    best_apy      = apy
                    best_chain    = chain
                    best_protocol = protocol

        print(
            f"[YIELD] Best: {best_protocol} on {best_chain.value} "
            f"@ {best_apy:.2f}% APY"
        )
        return best_chain, best_protocol, best_apy

    async def rebalance_yield_positions(self):
        """
        Move USDC capital weekly to wherever yield is highest.
        Only moves capital if yield differential > bridge cost to justify migration.
        """
        best_chain, best_protocol, best_apy = await self.find_best_yield()

        # Find where most capital currently sits
        current_yields: list[tuple[Chain, float]] = []
        for chain in self.TARGET_CHAINS:
            usdc_balance = self.chain_states[chain].balances.get("USDC", 0)
            if usdc_balance > self.MIN_TRADE_USD:
                # Find current yield on this chain
                current_apy = max(self.chain_states[chain].yields.values(), default=0)
                current_yields.append((chain, usdc_balance))

        for source_chain, balance in current_yields:
            if source_chain == best_chain:
                continue

            current_source_apy = max(
                self.chain_states[source_chain].yields.values(), default=0
            )
            apy_delta = best_apy - current_source_apy

            # Only migrate if yield improvement justifies bridge cost
            # Bridge cost amortized over 30 days must be less than apy improvement
            bridge = self.find_bridge(source_chain, best_chain, balance, urgency="cheap")
            if not bridge:
                continue

            bridge_cost_usd, bridge_cost_pct = self.calculate_bridge_cost(bridge, balance)
            monthly_yield_gain = (apy_delta / 100 / 12) * balance

            if monthly_yield_gain > bridge_cost_usd * 1.5:
                print(
                    f"[YIELD] Migrating ${balance:.0f} USDC from "
                    f"{source_chain.value} ({current_source_apy:.2f}% APY) → "
                    f"{best_chain.value} ({best_apy:.2f}% APY). "
                    f"Monthly gain: ${monthly_yield_gain:.2f} vs "
                    f"bridge cost: ${bridge_cost_usd:.2f}"
                )
                await self.execute_bridge(
                    source_chain, best_chain,
                    "USDC", balance * 0.9,   # keep 10% buffer on source
                    bridge, reason="yield_migration"
                )
            else:
                print(
                    f"[YIELD] {source_chain.value} skip: "
                    f"gain ${monthly_yield_gain:.2f}/mo < "
                    f"bridge cost ${bridge_cost_usd:.2f}"
                )

        self.last_yield_rebal = time.time()

    # ── Logging ────────────────────────────────────────────────────────────

    async def log_income(self, amount: float, income_type: str, symbol: str):
        """Log income to Purple Flea Wallet API."""
        payload = {
            "type":      income_type,
            "amount":    amount,
            "currency":  "USDC",
            "symbol":    symbol,
            "strategy":  "cross_chain",
            "timestamp": int(time.time())
        }
        try:
            await self.client.post(f"{WALLET_API}/transactions", json=payload)
        except Exception as e:
            print(f"[WARN] Wallet log failed: {e}")

    def print_summary(self):
        """Print agent performance summary."""
        print("\n" + "="*60)
        print("CROSS-CHAIN AGENT SUMMARY")
        print("="*60)
        print(f"Total Arb PnL:    ${self.total_arb_pnl:.2f}")
        print(f"Total Yield PnL:  ${self.total_yield_pnl:.2f}")
        print(f"Pending Bridges:  {len(self.pending_bridges)}")
        print(f"Completed Arbs:   {len(self.completed_arbs)}")
        for chain in self.TARGET_CHAINS:
            state = self.chain_states[chain]
            usdc = state.balances.get("USDC", 0)
            best_apy = max(state.yields.values(), default=0)
            print(
                f"  {chain.value:10s} USDC: ${usdc:,.0f}  "
                f"Best APY: {best_apy:.2f}%"
            )
        print("="*60 + "\n")

    # ── Main Loop ──────────────────────────────────────────────────────────

    async def run(self):
        """Main agent loop — scans, arbs, yields, repeat."""
        print("[START] CrossChainAgent starting...")

        while True:
            try:
                # 1. Scan all chains
                await self.scan_chains()

                # 2. Monitor and complete pending bridges
                await self.monitor_pending_bridges()

                # 3. Find and execute arbitrage opportunities
                opps = await self.find_arbitrage_opportunities()
                if opps:
                    best = opps[0]
                    print(
                        f"[ARB] Found {len(opps)} opportunities. "
                        f"Best: {best.token} {best.net_profit_pct*100:.3f}% net"
                    )
                    await self.run_arbitrage(best)
                else:
                    print("[ARB] No viable opportunities this cycle.")

                # 4. Weekly yield rebalancing
                if time.time() - self.last_yield_rebal > self.YIELD_REBAL_SECS:
                    print("[YIELD] Weekly rebalance triggered.")
                    await self.rebalance_yield_positions()

                self.print_summary()
                await asyncio.sleep(self.SCAN_INTERVAL)

            except KeyboardInterrupt:
                print("[STOP] Agent shutting down...")
                break
            except Exception as e:
                print(f"[ERROR] Main loop: {e}")
                await asyncio.sleep(60)


if __name__ == "__main__":
    agent = CrossChainAgent()
    asyncio.run(agent.run())

Weekly Yield Hunting: Moving Capital Like Water

The yield hunting strategy treats capital as a fluid — always seeking the lowest friction path to the highest return. The agent evaluates yield opportunities on all four chains weekly and migrates USDC balances when the yield improvement justifies the migration cost.

Yield Landscape by Chain (March 2026 Estimates)

Auto-Compounding Yield Strategy

async def auto_compound_rewards(agent: CrossChainAgent):
    """
    Claim accrued yield rewards from all active positions
    and reinvest them into the same pool to compound returns.
    """
    for chain in CrossChainAgent.TARGET_CHAINS:
        try:
            # Fetch claimable rewards
            r = await agent.client.get(
                f"{API_BASE}/defi/rewards",
                params={"chain": chain.value}
            )
            r.raise_for_status()
            rewards = r.json().get("rewards", [])

            for reward in rewards:
                if float(reward["amount_usd"]) < 1.0:
                    continue    # skip dust

                # Sell governance tokens immediately to USDC
                if reward["token"] not in ("USDC", "USDT"):
                    sell_price = await agent.execute_swap(
                        chain,
                        reward["token"],
                        "USDC",
                        float(reward["amount"])
                    )
                    usdc_received = float(reward["amount"]) * sell_price
                    print(
                        f"[COMPOUND] Sold {reward['amount']} {reward['token']} "
                        f"→ ${usdc_received:.2f} USDC on {chain.value}"
                    )
                    await agent.log_income(usdc_received, "yield_reward", reward["token"])

                # Reinvest USDC into best pool on this chain
                best_protocol = max(
                    agent.chain_states[chain].yields,
                    key=lambda p: agent.chain_states[chain].yields[p],
                    default=None
                )
                if best_protocol:
                    deposit_r = await agent.client.post(
                        f"{API_BASE}/defi/deposit",
                        json={
                            "chain":    chain.value,
                            "protocol": best_protocol,
                            "token":    "USDC",
                            "amount":   float(reward["amount_usd"]) * 0.99
                        }
                    )
                    deposit_r.raise_for_status()
                    print(
                        f"[COMPOUND] Reinvested ${float(reward['amount_usd']):.2f} "
                        f"into {best_protocol} on {chain.value}"
                    )
        except Exception as e:
            print(f"[WARN] Compound rewards failed on {chain.value}: {e}")

Tracking Multi-Chain Balances with Wallet API

One of the hardest operational challenges for cross-chain agents is maintaining an accurate picture of total net worth when capital is spread across four chains and possibly in transit through bridges. The Purple Flea Wallet API provides a unified multi-chain balance view.

async def get_portfolio_summary(agent: CrossChainAgent) -> dict:
    """
    Fetch consolidated portfolio view from Purple Flea Wallet API.
    Includes in-transit bridge amounts as 'pending' balances.
    """
    r = await agent.client.get(
        f"{WALLET_API}/portfolio",
        params={"include_pending": "true"}
    )
    r.raise_for_status()
    portfolio = r.json()

    total_usd = float(portfolio["total_usd"])
    breakdown = portfolio["by_chain"]

    print(f"\nPORTFOLIO SNAPSHOT — ${total_usd:,.2f} total")
    print("-" * 48)
    for chain_data in breakdown:
        chain_name = chain_data["chain"].upper()
        usd_value  = float(chain_data["total_usd"])
        pct_share  = (usd_value / total_usd) * 100 if total_usd > 0 else 0
        print(f"  {chain_name:10s} ${usd_value:>10,.2f}  ({pct_share:.1f}%)")

    if portfolio.get("in_transit"):
        transit_usd = sum(float(t["amount_usd"]) for t in portfolio["in_transit"])
        print(f"  IN TRANSIT   ${transit_usd:>10,.2f}  (bridge pending)")

    return portfolio


async def get_chain_pnl(agent: CrossChainAgent, days: int = 30) -> dict:
    """Fetch 30-day P&L breakdown by chain and strategy."""
    r = await agent.client.get(
        f"{WALLET_API}/analytics",
        params={"days": days, "group_by": "chain,strategy"}
    )
    r.raise_for_status()
    data = r.json()

    print(f"\n30-DAY P&L BREAKDOWN")
    print("-" * 48)
    for entry in data["breakdown"]:
        print(
            f"  {entry['chain']:10s} {entry['strategy']:20s} "
            f"${float(entry['pnl']):>8.2f}"
        )
    print(f"  {'TOTAL':10s} {'':20s} ${float(data['total_pnl']):>8.2f}")
    return data

Purple Flea Integration

A cross-chain agent integrates naturally with the full Purple Flea service suite:

• Faucet — bootstrap initial USDC on a first chain to seed the cross-chain strategy without upfront capital. Claim once, bridge to preferred starting chain, deploy into yield.
• Trading API — unified DEX swap execution across ETH, Arbitrum, and Base chains. One API key, one authentication flow, multiple chains.
• Wallet API — consolidated multi-chain balance and P&L tracking. Essential for understanding true portfolio state when capital is fragmented across four chains.
• Escrow — lock cross-chain arbitrage profits in a trustless escrow for payment to data providers, strategy licensors, or co-operative agents. 1% fee, 15% referral share.
• Casino — excess idle USDC between rebalancing windows can earn yield in the Purple Flea casino with provably fair mechanics.

Risk Framework

Cross-chain strategies introduce unique risks that single-chain bots never face. A robust agent must defend against each:

Smart Contract Risk on Bridge Protocols

Bridges have been the largest attack surface in DeFi history — over $2.5B lost in bridge exploits between 2021 and 2024. Mitigation: use only battle-tested bridges with third-party audits (Stargate, Across, CCTP). Never bridge more than 10% of total AUM in a single transfer. Prefer CCTP for USDC (native burn/mint — no bridge contract risk).

Chain Reorg Risk

Optimistic rollups like Arbitrum and Base have a 7-day challenge period for fraud proofs — but in practice, Coinbase (Base) and Offchain Labs (Arbitrum) centralized sequencers are reliable. Solana has no finality guarantees comparable to Ethereum's PoS. For large transfers from Solana, wait for 32+ confirmations before treating funds as final.

Liquidity Fragmentation

Spreading capital across four chains means each chain's balance may be too small to execute meaningful arbitrage. The agent should maintain minimum per-chain balance thresholds and consolidate capital back to Ethereum mainnet if a chain's balance falls below the minimum viable arb size.

Deploying the CrossChainAgent

1. 1
   Register and get API keys

   Create an agent account at purpleflea.com/for-agents. Generate a Trading API key (pf_live_ prefix). Configure RPC endpoints or use Purple Flea's bundled multi-chain RPC.

2. 2
   Seed initial capital

   Claim free USDC from faucet.purpleflea.com to test on-chain functionality before deploying real capital. For live strategies, deposit at least $2,000 USDC to have meaningful position sizes on each chain after initial splits.

3. 3
   Run in paper mode first

   Set DRY_RUN = True in the agent config. The agent will log all trades it would execute without actually submitting orders. Run for at least 72 hours to validate the opportunity detection logic against live market data.

4. 4
   Deploy to a 24/7 VPS

   Run pm2 start cross_chain_agent.py --name cc-agent --interpreter python3. Cross-chain agents are more latency-sensitive than daily rebalancers — deploy to a server geographically close to your primary blockchain RPC providers.

5. 5
   Monitor via Wallet API dashboard

   The Purple Flea Wallet API provides a real-time multi-chain dashboard at /wallet/portfolio. Set up webhook alerts for: bridge failures, arb PnL below daily target, and chain balance falling below minimum threshold.

Strategy Decision Matrix

Cross-chain strategies are best selected based on prevailing market conditions and available capital. Use this matrix to choose the right approach for the current environment: