DeFi Aggregators for AI Agents: 1inch, ParaSwap, and Optimal Swap Routing

DEX aggregators have become the de facto standard for efficient on-chain token swaps. Rather than routing through a single liquidity source, aggregators like 1inch and ParaSwap split orders across dozens of pools to find the best price, minimize slippage, and—increasingly—protect against MEV extraction. For AI agents operating autonomously, integrating aggregators programmatically is not optional: the spread difference between a naive direct swap and an aggregator-routed swap can exceed 0.5% on medium-sized orders, which compounds catastrophically over thousands of transactions.

This guide covers the mechanics of how aggregators work, how AI agents should interact with them, MEV protection options, and a complete Python implementation that queries multiple aggregators and selects the optimal route—integrating with the Purple Flea Wallet API for execution.

How DEX Aggregators Work

At the core of every DEX aggregator is a pathfinding algorithm. The algorithm treats liquidity pools as nodes in a graph and token swap steps as edges. Given a source token, destination token, and amount, the algorithm explores all paths (direct swaps, two-hop routes, three-hop routes) and returns the path—or split of paths—that maximizes output.

Split Routing

Split routing is the key innovation that separates aggregators from simple DEX routers. When a swap order is large enough to move the price of a single pool, splitting the order across multiple pools achieves a better average execution price. For example, a 100 ETH → USDC swap might be split:

• 40% through Uniswap V3 (concentrated ETH/USDC pool at 0.05% fee tier)
• 35% through Curve (3pool route via USDT bridge)
• 25% through Balancer (weighted pool with lower slippage for large orders)

Each sub-order moves the price of its pool less than a single-venue trade would, and the weighted average execution price across all pools is better than any single pool could offer alone.

Gas Optimization

Split routing has a cost: each additional pool hop adds gas. A well-designed aggregator balances price improvement against gas cost. The aggregator's output amount calculation must account for gas in the comparison:

net_output = raw_output_amount - (gas_cost_in_gwei * gas_price * eth_price / 1e18)

In practice, aggregators maintain a gas cost model for each protocol and weigh whether a marginal route improvement justifies the extra gas. For small swaps (<$500), a simple direct Uniswap V3 route often wins on net output because the gas overhead of split routing exceeds the price improvement. For swaps >$10,000, split routing almost always wins.

AI agents should always request net output after gas from the aggregator API rather than raw output. Both 1inch and ParaSwap expose this in their quote responses.

MEV Protection

MEV (Maximal Extractable Value) bots monitor the public mempool and can front-run large swap transactions. When your agent broadcasts a 50 ETH swap at market price, a bot may detect the transaction, insert its own buy order before yours, and sell immediately after—extracting value from your trade's price impact. This is called a sandwich attack.

Modern aggregators address MEV through several mechanisms:

Mechanism	How It Works	Trade-off
Private mempool / Flashbots	Sends txn directly to block builders, bypassing public mempool	Requires bundler integration, higher complexity
Tight slippage + revert	Transaction reverts if price moves beyond threshold	May cause failed txns on volatile assets
Commit-reveal	Intent is hidden until execution block	Adds latency (2 blocks minimum)
RFQ / Fusion mode	Resolvers compete off-chain; winning resolver executes atomically	Resolver availability required

1inch Fusion Mode

1inch Fusion is the protocol's intent-based order system. Instead of broadcasting a transaction directly, your agent signs an intent (EIP-712 typed data) specifying: input token, output token, minimum output amount, expiry, and optional partial fill constraints. A network of resolvers compete to fill the order—they can aggregate their own liquidity sources, batch multiple orders together, and use private block builder channels.

The key advantage for AI agents: no gas is paid by the trader. Resolvers pay gas and profit from any arbitrage between the stated minimum price and the market price. This means your agent's Ethereum wallet does not need ETH for gas—only the ERC-20 token being sold.

Fusion is available via the 1inch Dev Portal API (https://api.1inch.dev). The flow is:

1. Request a quote: GET /swap/v6.0/1/quote
2. Generate an order: GET /orderbook/v4.0/1/order/quote
3. Sign the EIP-712 order with your agent's private key
4. Submit to the resolver network: POST /orderbook/v4.0/1/order
5. Poll order status until filled or expired

ParaSwap Delta

ParaSwap Delta is ParaSwap's equivalent intent-based system. Delta aggregates ParaSwap's liquidity graph with off-chain market maker quotes (RFQ), resulting in particularly strong pricing for stablecoin pairs and high-volume tokens. Delta orders are executed by market makers who have pre-signed liquidity commitments on-chain.

Delta differs from Fusion in that execution is guaranteed within a defined time window (typically 30 seconds) if the price hasn't moved beyond the stated minimum. This predictability is valuable for agents that need to sequence multiple operations and cannot tolerate order fill uncertainty.

The ParaSwap Delta API flow:

1. Get a price quote: GET https://apiv5.paraswap.io/prices?srcToken=...&destToken=...&amount=...&network=1
2. Build a transaction: POST https://apiv5.paraswap.io/transactions/1
3. Sign and submit

Slippage Comparison Across Aggregators

Slippage is not equal across aggregators for all token pairs. For major pairs (ETH/USDC, WBTC/USDT), 1inch and ParaSwap are within 0.01-0.03% of each other on most order sizes. The divergence appears at the edges:

Pair / Size	Uniswap V3 Direct	1inch	ParaSwap	0x API
ETH/USDC $5K	0.08%	0.06%	0.07%	0.07%
ETH/USDC $50K	0.41%	0.18%	0.19%	0.22%
ETH/USDC $500K	2.8%	0.72%	0.78%	0.91%
Long-tail ERC-20 $10K	0.9%	0.65%	0.80%	0.75%

For AI agents making frequent medium-to-large swaps, the choice of aggregator can meaningfully affect P&L. The best practice is to query multiple aggregators and pick the best net output for each trade rather than hardcoding a single aggregator.

Purple Flea Wallet API and Routing

The Purple Flea Wallet API provides a unified interface for multi-chain asset management. For EVM chains (ETH, BNB, MATIC), the API uses wagyu.xyz for swap routing—a meta-aggregator that queries 1inch, ParaSwap, 0x, and CowSwap simultaneously and returns the best route. This means agents using the Purple Flea wallet API get aggregator-level pricing without integrating each aggregator separately.

The wallet API endpoint for swap quotes:

GET https://purpleflea.com/api/wallet/swap/quote
  ?from_token=ETH
  &to_token=USDC
  &amount=1.5
  &chain=ethereum
  &agent_id=YOUR_AGENT_ID

Response includes route_details showing which underlying aggregators were queried and the winning route's composition.

Python Agent: Multi-Aggregator Route Comparison

The following Python agent queries both 1inch and ParaSwap for a given swap, compares net outputs, and executes through the better route via Purple Flea's wallet API.

"""
Multi-aggregator swap router for AI agents.
Queries 1inch and ParaSwap, selects best net output,
and executes via Purple Flea Wallet API.
"""

import asyncio
import aiohttp
import json
from dataclasses import dataclass
from typing import Optional

# Configuration
PURPLE_FLEA_API = "https://purpleflea.com/api"
ONEINCH_API = "https://api.1inch.dev/swap/v6.0/1"
PARASWAP_API = "https://apiv5.paraswap.io"

ONEINCH_API_KEY = "YOUR_1INCH_API_KEY"
PF_AGENT_ID = "YOUR_AGENT_ID"
PF_API_KEY = "YOUR_PF_API_KEY"

# Token addresses (Ethereum mainnet)
TOKENS = {
    "ETH":  "0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE",
    "USDC": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
    "WBTC": "0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599",
    "USDT": "0xdAC17F958D2ee523a2206206994597C13D831ec7",
}

ETH_PRICE_USD = 3_200.0  # fetch live in production
GAS_PRICE_GWEI = 12.0


@dataclass
class QuoteResult:
    aggregator: str
    src_amount: int
    dst_amount: int
    gas_estimate: int
    net_dst_amount: float  # after gas cost deducted in dst token terms
    route_summary: str


async def get_1inch_quote(
    session: aiohttp.ClientSession,
    src_token: str,
    dst_token: str,
    amount_wei: int
) -> Optional[QuoteResult]:
    """Fetch a swap quote from 1inch API."""
    url = f"{ONEINCH_API}/quote"
    params = {
        "src": TOKENS[src_token],
        "dst": TOKENS[dst_token],
        "amount": str(amount_wei),
        "includeGas": "true",
    }
    headers = {"Authorization": f"Bearer {ONEINCH_API_KEY}"}
    try:
        async with session.get(url, params=params, headers=headers) as resp:
            if resp.status != 200:
                print(f"1inch quote error: {resp.status}")
                return None
            data = await resp.json()
            dst_amount = int(data["dstAmount"])
            gas = int(data.get("gas", 200_000))
            # Gas cost in USD, then convert to dst token units
            gas_cost_usd = gas * GAS_PRICE_GWEI * 1e-9 * ETH_PRICE_USD
            # Assume dst is USDC (6 decimals) for simplicity
            gas_cost_dst = gas_cost_usd * 1e6
            net = dst_amount - gas_cost_dst
            protocols = data.get("protocols", [])
            summary = _summarize_protocols(protocols)
            return QuoteResult(
                aggregator="1inch",
                src_amount=amount_wei,
                dst_amount=dst_amount,
                gas_estimate=gas,
                net_dst_amount=net,
                route_summary=summary,
            )
    except Exception as e:
        print(f"1inch quote exception: {e}")
        return None


async def get_paraswap_quote(
    session: aiohttp.ClientSession,
    src_token: str,
    dst_token: str,
    amount_wei: int
) -> Optional[QuoteResult]:
    """Fetch a swap quote from ParaSwap API."""
    url = f"{PARASWAP_API}/prices"
    params = {
        "srcToken": TOKENS[src_token],
        "destToken": TOKENS[dst_token],
        "amount": str(amount_wei),
        "srcDecimals": "18" if src_token == "ETH" else "6",
        "destDecimals": "6",  # assume USDC
        "side": "SELL",
        "network": "1",
        "includeDEXS": "UniswapV3,Curve,Balancer,SushiSwap",
    }
    try:
        async with session.get(url, params=params) as resp:
            if resp.status != 200:
                print(f"ParaSwap quote error: {resp.status}")
                return None
            data = await resp.json()
            price_route = data.get("priceRoute", {})
            dst_amount = int(price_route.get("destAmount", 0))
            gas = int(price_route.get("gasCost", 200_000))
            gas_cost_usd = gas * GAS_PRICE_GWEI * 1e-9 * ETH_PRICE_USD
            gas_cost_dst = gas_cost_usd * 1e6
            net = dst_amount - gas_cost_dst
            # Build route summary from bestRoute
            best_route = price_route.get("bestRoute", [])
            pools = []
            for leg in best_route:
                for swap in leg.get("swaps", []):
                    for ex in swap.get("swapExchanges", []):
                        pools.append(f"{ex['exchange']}({ex['percent']}%)")
            summary = " + ".join(pools) if pools else "direct"
            return QuoteResult(
                aggregator="ParaSwap",
                src_amount=amount_wei,
                dst_amount=dst_amount,
                gas_estimate=gas,
                net_dst_amount=net,
                route_summary=summary,
            )
    except Exception as e:
        print(f"ParaSwap quote exception: {e}")
        return None


def _summarize_protocols(protocols: list) -> str:
    """Extract protocol names from 1inch protocol list."""
    names = set()
    for outer in protocols:
        for inner in outer:
            for part in inner:
                names.add(part.get("name", "unknown"))
    return ", ".join(sorted(names)) if names else "direct"


async def find_best_route(
    src_token: str,
    dst_token: str,
    amount_wei: int
) -> Optional[QuoteResult]:
    """Query all aggregators and return the best quote by net output."""
    async with aiohttp.ClientSession() as session:
        quotes = await asyncio.gather(
            get_1inch_quote(session, src_token, dst_token, amount_wei),
            get_paraswap_quote(session, src_token, dst_token, amount_wei),
        )
    valid = [q for q in quotes if q is not None]
    if not valid:
        return None
    best = max(valid, key=lambda q: q.net_dst_amount)
    print("\n--- Aggregator Comparison ---")
    for q in valid:
        marker = " <-- BEST" if q is best else ""
        print(
            f"  {q.aggregator}: dst={q.dst_amount/1e6:.4f} USDC "
            f"net={q.net_dst_amount/1e6:.4f} USDC "
            f"gas={q.gas_estimate:,}{marker}"
        )
        print(f"    Route: {q.route_summary}")
    print()
    return best


async def execute_swap_via_purple_flea(
    best: QuoteResult,
    src_token: str,
    dst_token: str
) -> dict:
    """Execute the selected swap via Purple Flea Wallet API."""
    payload = {
        "agent_id": PF_AGENT_ID,
        "from_token": src_token,
        "to_token": dst_token,
        "amount": str(best.src_amount),
        "preferred_aggregator": best.aggregator.lower().replace("paraswap", "paraswap"),
        "min_output": str(int(best.dst_amount * 0.995)),  # 0.5% slippage tolerance
        "chain": "ethereum",
    }
    headers = {
        "Authorization": f"Bearer {PF_API_KEY}",
        "Content-Type": "application/json",
    }
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{PURPLE_FLEA_API}/wallet/swap/execute",
            json=payload,
            headers=headers
        ) as resp:
            result = await resp.json()
            return result


async def main():
    # Example: swap 1.5 ETH to USDC
    src = "ETH"
    dst = "USDC"
    amount_eth = 1.5
    amount_wei = int(amount_eth * 1e18)

    print(f"Finding best route for {amount_eth} {src} -> {dst}...")
    best = await find_best_route(src, dst, amount_wei)
    if best is None:
        print("No valid quotes received. Aborting.")
        return

    print(f"Best route: {best.aggregator}")
    print(f"Expected output: {best.dst_amount / 1e6:.4f} USDC")
    print(f"Net after gas:   {best.net_dst_amount / 1e6:.4f} USDC")

    # Uncomment to execute:
    # result = await execute_swap_via_purple_flea(best, src, dst)
    # print("Execution result:", json.dumps(result, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

MEV Protection Strategy for Autonomous Agents

For agents executing swaps autonomously, MEV is a persistent threat. The recommended strategy combines multiple layers:

Private RPC Endpoints

Use a private RPC (Flashbots Protect, MEV Blocker, or Beaverbuild's private endpoint) for transaction submission. This routes your transaction directly to a block builder rather than the public mempool, eliminating the most common sandwich attack vector. Set your RPC as:

PRIVATE_RPC = "https://rpc.flashbots.net"
# or
PRIVATE_RPC = "https://mev-blocker.io"
# or for BNB Chain:
PRIVATE_RPC = "https://api.48.club"

Fusion Mode for Large Orders

For orders above $25,000, use 1inch Fusion or ParaSwap Delta instead of direct swap transactions. The intent-based model means your agent never broadcasts a swap transaction that can be front-run—resolvers receive the signed intent and execute it privately.

Dynamic Slippage Settings

Slippage tolerance should adapt to market conditions. Tight slippage reduces MEV exposure but increases the chance of a failed transaction during volatile periods. A reasonable heuristic for AI agents:

def compute_slippage(volatility_1h: float, order_size_usd: float) -> float:
    """
    Compute adaptive slippage tolerance.
    volatility_1h: 1-hour price volatility (e.g., 0.02 = 2%)
    order_size_usd: order size in USD
    """
    base = 0.003  # 0.3% minimum
    vol_component = volatility_1h * 0.5  # scale with market vol
    # Larger orders need more slippage tolerance for partial fills
    size_component = min(order_size_usd / 1_000_000 * 0.01, 0.005)
    return min(base + vol_component + size_component, 0.02)  # cap at 2%

Cross-Chain Routing Considerations

AI agents operating across multiple chains (ETH, BNB, MATIC, Arbitrum, Base) need to account for bridging costs in addition to swap costs. The Purple Flea wallet supports ETH, BNB, and MATIC natively, allowing agents to hold and swap assets across chains through a single API without managing separate wallets and RPC connections per chain.

When planning a cross-chain swap (e.g., USDC on Ethereum to USDC on Arbitrum), the total cost includes:

• Bridge fee (typically 0.01-0.05% for official bridges, higher for third-party)
• Gas on source chain (ETH mainnet can be expensive)
• Gas on destination chain (Arbitrum/Base much cheaper)
• Bridge finality time (official bridges: 7 days for Optimistic, ~20 minutes for ZK)

For time-sensitive agents, using a third-party bridge aggregator (Li.Fi, Socket) or sticking to same-chain operations when possible minimizes latency and cost.

Monitoring and Alerting

An autonomous swap agent should monitor its own performance and alert on anomalies:

import statistics

class SwapMonitor:
    def __init__(self, window_size: int = 100):
        self.slippage_history = []
        self.window_size = window_size

    def record_swap(self, quoted_output: float, actual_output: float):
        realized_slippage = (quoted_output - actual_output) / quoted_output
        self.slippage_history.append(realized_slippage)
        if len(self.slippage_history) > self.window_size:
            self.slippage_history.pop(0)

    def check_anomaly(self) -> bool:
        if len(self.slippage_history) < 10:
            return False
        mean = statistics.mean(self.slippage_history)
        stdev = statistics.stdev(self.slippage_history)
        latest = self.slippage_history[-1]
        # Alert if latest slippage is >3 std deviations above mean
        if latest > mean + 3 * stdev:
            print(f"ALERT: Unusual slippage detected: {latest:.4%} "
                  f"(mean: {mean:.4%}, stdev: {stdev:.4%})")
            return True
        return False

    def average_slippage(self) -> float:
        return statistics.mean(self.slippage_history) if self.slippage_history else 0.0

Gas Token Strategies

One often-overlooked aspect of agent swap optimization is gas token management. Agents that execute many swaps on ETH mainnet need a reliable supply of ETH for gas. Strategies include:

• ETH reserve policy: Maintain a minimum ETH balance (e.g., 0.1 ETH) and automatically buy ETH if balance drops below threshold
• Use Fusion mode: 1inch Fusion eliminates gas costs for the trader—resolvers pay gas
• Batch operations: Use multicall contracts (e.g., via Gnosis Safe) to batch multiple swaps into a single transaction
• Off-peak timing: Monitor gas prices and defer non-urgent swaps to low-gas periods (typically UTC 02:00-08:00)

Summary

DEX aggregators are essential infrastructure for AI agents that interact with on-chain markets. Key takeaways:

• Always query multiple aggregators and pick the best net output after gas, not just raw output
• Use intent-based systems (1inch Fusion, ParaSwap Delta) for large orders to avoid MEV
• Route through private RPC endpoints to eliminate mempool-based sandwich attacks
• Implement adaptive slippage that responds to market volatility
• Monitor realized slippage against quoted output to detect execution anomalies
• Use the Purple Flea Wallet API for unified multi-chain swap access with built-in aggregator routing

The Python examples above provide a starting framework. In production, extend with database logging of all swaps, per-token slippage history, gas price feeds, and circuit breakers that halt trading when market conditions deviate from norms.

Aggregator Risk Management

DEX aggregators introduce additional risk vectors that AI agents must account for:

Smart Contract Risk

Aggregator contracts are complex—they interact with dozens of underlying protocols in a single transaction. Smart contract bugs in aggregator router logic have caused losses in the past. Mitigations for agents:

• Set a per-swap maximum size limit to bound smart contract risk exposure
• Prefer aggregators that have undergone multiple security audits (1inch, ParaSwap, 0x all have extensive audit histories)
• Use permit2 or ERC-20 allowance revocation between swaps to avoid unlimited approvals sitting on-chain
• Monitor aggregator protocol security feeds and pause swapping if a vulnerability is disclosed

Oracle Manipulation Risk

Some aggregator routes pass through pools that use on-chain price oracles. If an attacker can manipulate the oracle (flash loan attack), they can make the aggregator route through a manipulated pool. For large swaps, verify the route's pool sources and avoid routing through low-liquidity or oracle-dependent pools for the final pricing step.

API Dependency and Fallback

Aggregator APIs can experience downtime. The Python agent above handles this via the Optional[QuoteResult] return—if an API fails, it returns None and the router falls back to remaining aggregators. For production agents, maintain a local fallback: a direct Uniswap V3 quote that does not depend on any aggregator API:

async def get_uniswap_v3_fallback_quote(
    session: aiohttp.ClientSession,
    src_token: str,
    dst_token: str,
    amount_wei: int,
    rpc_url: str = "https://eth.llamarpc.com",
) -> Optional[QuoteResult]:
    """
    Direct on-chain Uniswap V3 quote as aggregator fallback.
    Uses eth_call to the QuoterV2 contract.
    """
    QUOTER_V2 = "0x61fFE014bA17989E743c5F6cB21bF9697530B21e"
    # ABI encode: quoteExactInputSingle(tokenIn, tokenOut, fee, amountIn, sqrtPriceLimitX96)
    # For brevity, using a simplified direct approach
    # In production, use web3.py or ethers.js with the full ABI
    payload = {
        "jsonrpc": "2.0",
        "method": "eth_call",
        "params": [{
            "to": QUOTER_V2,
            # Encoded call data would go here
            "data": "0x..."  # Encode quoteExactInputSingle
        }, "latest"],
        "id": 1,
    }
    try:
        async with session.post(rpc_url, json=payload) as resp:
            data = await resp.json()
            result_hex = data.get("result", "0x0")
            dst_amount = int(result_hex, 16)
            return QuoteResult(
                aggregator="UniswapV3-Direct",
                src_amount=amount_wei,
                dst_amount=dst_amount,
                gas_estimate=180_000,
                net_dst_amount=float(dst_amount),
                route_summary="UniswapV3 direct (0.05% pool)",
            )
    except Exception as e:
        print(f"UniswapV3 fallback failed: {e}")
        return None

Multi-Chain Aggregator Landscape

The aggregator landscape differs substantially by chain. Agents operating on non-Ethereum chains need chain-specific aggregator integrations:

Chain	Primary Aggregator	Secondary	API Style
Ethereum	1inch	ParaSwap, 0x	REST API
BNB Chain	1inch (BNB)	PancakeSwap Router	REST API
Polygon/MATIC	1inch (137)	ParaSwap (137)	REST API
Arbitrum	1inch (42161)	Camelot Aggregator	REST API
Solana	Jupiter	Raydium	REST API
Base	1inch (8453)	BaseSwap	REST API

For 1inch and ParaSwap, the API structure is identical across chains—only the chain ID parameter changes. This makes multi-chain integration straightforward: parameterize the chain ID and run the same quote comparison logic across all supported chains.

Jupiter is the dominant aggregator on Solana with a REST API at https://quote-api.jup.ag/v6/quote. For agents using the Purple Flea Wallet's Solana support, Jupiter routing is used automatically under the hood.

Performance Benchmarks for Agent Swap Loops

An AI agent that executes many swaps per day needs to understand the performance characteristics of the aggregator query loop. Typical latencies for the full quote-and-execute cycle:

• 1inch API quote: 150-400ms average; spikes to 1-2s during high congestion
• ParaSwap API quote: 100-350ms average
• Parallel query (both APIs): max(1inch_latency, paraswap_latency) — typically 200-400ms total
• Transaction broadcast to mempool: 50-100ms
• Block confirmation (ETH L1): 12 seconds average, 24 seconds for 99th percentile
• Full cycle (quote + execute + confirm): ~15-30 seconds on ETH L1

For agents that need sub-second swap execution (e.g., arbitrage bots), direct contract interaction is required—aggregator APIs add too much latency for time-critical strategies. For agents making strategic swaps (rebalancing, harvesting, position sizing), the aggregator latency is well within acceptable bounds.

Cache aggregator quotes for up to 30 seconds if re-using them for multiple decisions in the same cycle. Prices move slowly enough that a 30-second-old quote is usually still valid for non-time-critical swaps, saving API call overhead.

Long-Tail Token Routing

The aggregator comparison tables above focus on major tokens (ETH, WBTC, USDC). For long-tail tokens with limited liquidity, aggregator routing becomes even more critical—and more complex.

Long-tail token characteristics that affect routing:

• Limited pool count: Many long-tail tokens exist in only one or two pools, making split routing impossible or ineffective
• High bid-ask spread: Thin liquidity creates wide spreads; a 2-3% slippage is common even for modest order sizes
• Single hop vs. multi-hop: Most aggregators route long-tail tokens through an intermediate (ETH or USDC) to reach the destination token
• Curve pools: For stablecoin-pegged assets (LSTs, yield tokens), Curve often has the best pricing even for tokens not listed on major DEXes

Agents should apply stricter slippage limits when routing long-tail tokens and always verify that the output amount is within a reasonable range of the mid-market price before execution:

async def sanity_check_quote(
    quote: QuoteResult,
    src_amount_usd: float,
    mid_market_price_usd: float,  # external oracle price
    max_deviation_pct: float = 0.05,  # reject if >5% below mid-market
) -> bool:
    """
    Sanity check: reject quotes that are too far from mid-market price.
    Protects against stale aggregator quotes and oracle manipulation.
    """
    quoted_output_usd = quote.dst_amount / 1e6  # assuming USDC 6 decimals
    if mid_market_price_usd <= 0:
        return True  # no oracle available, skip check

    expected_usd = src_amount_usd  # for token->USDC swaps
    deviation = (expected_usd - quoted_output_usd) / expected_usd

    if deviation > max_deviation_pct:
        print(
            f"SANITY CHECK FAILED: quoted output {quoted_output_usd:.2f} USDC "
            f"is {deviation:.1%} below expected {expected_usd:.2f} USDC "
            f"(max allowed: {max_deviation_pct:.1%})"
        )
        return False
    return True

Advanced Routing Patterns

Price-Impact Aware Order Sizing

Before sending a large order, query the aggregator's price impact estimate for multiple order sizes and identify the "break point" where price impact starts accelerating. Split the order into tranches timed apart to give the pool time to rebalance from arbitrageurs:

async def find_optimal_tranche_size(
    session: aiohttp.ClientSession,
    src_token: str,
    dst_token: str,
    total_amount_wei: int,
    test_sizes: List[float] = [0.1, 0.2, 0.3, 0.5, 0.75, 1.0],
) -> float:
    """
    Find the tranche size (as fraction of total) that minimizes price impact.
    Returns optimal fraction for a single tranche.
    """
    results = []
    for frac in test_sizes:
        tranche_wei = int(total_amount_wei * frac)
        # Extrapolate: total cost if we send 1/frac tranches of this size
        quote = await get_1inch_quote(session, src_token, dst_token, tranche_wei)
        if quote is None:
            continue
        # Rate: dst per src wei
        rate = quote.dst_amount / tranche_wei
        # Extrapolated total output if all tranches at this rate
        extrapolated_total = rate * total_amount_wei
        results.append((frac, extrapolated_total))

    if not results:
        return 1.0  # fallback: send all at once

    # Pick fraction that maximizes extrapolated total output
    best = max(results, key=lambda r: r[1])
    print(f"Optimal tranche size: {best[0]*100:.0f}% of total")
    return best[0]

Time-Weighted Execution (TWAP)

For very large orders, TWAP (Time-Weighted Average Price) execution spreads a large order over time to achieve the average market price rather than a single snapshot price. TWAP is commonly used in TradFi for institutional equity orders and is increasingly adopted in DeFi via protocols like TWAMM (Time-Weighted AMM).

A simple agent-level TWAP implementation:

async def execute_twap(
    src_token: str,
    dst_token: str,
    total_amount_wei: int,
    duration_minutes: int = 60,
    num_tranches: int = 12,
    execute_fn = None,  # async function(src, dst, amount_wei) -> bool
):
    """Execute a TWAP order: split total into equal tranches, spaced evenly."""
    tranche_size = total_amount_wei // num_tranches
    interval_sec = (duration_minutes * 60) / num_tranches
    print(
        f"TWAP: {num_tranches} tranches of {tranche_size/1e18:.4f} ETH "
        f"every {interval_sec:.0f}s over {duration_minutes}min"
    )
    for i in range(num_tranches):
        print(f"Tranche {i+1}/{num_tranches}")
        if execute_fn:
            await execute_fn(src_token, dst_token, tranche_size)
        if i < num_tranches - 1:
            await asyncio.sleep(interval_sec)
    print("TWAP complete.")

TWAP is most effective for orders above $100,000 where single-shot execution would move the market by more than 0.5%. For AI agents making routine portfolio rebalancing trades, TWAP with Purple Flea's wallet API provides institutional-grade execution without requiring a dedicated prime brokerage relationship.