AI Agents as Liquidity Providers: Passive Income from Protocol Markets

Decentralized exchanges generate billions in trading fees daily. Unlike human LPs who set positions and forget, AI agents can continuously monitor pool conditions, rebalance concentrated ranges before price escapes, collect fees the moment they accrue, and hedge impermanent loss in real time using perpetual futures. This guide covers the complete liquidity provision stack for autonomous agents.

Liquidity Provision Fundamentals

Automated market makers (AMMs) replaced traditional order books with a simple insight: instead of matching buyers and sellers, pool tokens in a smart contract and let a mathematical formula set the price. Liquidity providers deposit token pairs, earn a pro-rata share of every swap fee, and bear impermanent loss when the price ratio shifts.

The canonical constant-product formula is x * y = k, where x and y are the token reserves and k is a constant. Every swap moves the price along this curve. LPs earn fees proportional to their share of total liquidity, but suffer IL relative to simply holding the tokens.

Full-Range vs Concentrated Liquidity

Classic AMMs spread LP capital across all prices from zero to infinity. Uniswap V3 introduced concentrated liquidity: LPs choose a price range [P_lower, P_upper] and deploy capital only within that band. Capital efficiency can be 4000x higher in a tight range versus full-range, but the LP earns zero fees when price leaves the range.

Fee Income Calculation

Understanding the math behind fee income lets agents set realistic return expectations and compare positions objectively. The core formula:

Annualized Return (APR) from Fees

Concentrated Range Multiplier

When you deploy capital in a concentrated range that covers only a fraction of the full price curve, your effective liquidity multiplier is:

Python LiquidityManager Class

The following class provides a complete interface for an AI agent to manage LP positions. It handles position creation, fee collection, impermanent loss tracking, and auto-rebalancing. Integrate with the Purple Flea Trading API for hedging and the Wallet API for income tracking.

import math
import time
import requests
from dataclasses import dataclass, field
from typing import Optional, Dict, List, Tuple
from decimal import Decimal
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger('LiquidityManager')

PURPLE_FLEA_BASE = "https://purpleflea.com/api/v1"

@dataclass
class LPPosition:
    position_id: str
    pool_address: str
    token0: str
    token1: str
    fee_tier: float          # 0.0005, 0.003, or 0.01
    tick_lower: int
    tick_upper: int
    liquidity: int
    entry_price: float
    token0_deposited: float
    token1_deposited: float
    fees_collected_token0: float = 0.0
    fees_collected_token1: float = 0.0
    created_at: float = field(default_factory=time.time)
    last_collected: float = field(default_factory=time.time)
    rebalance_count: int = 0


class LiquidityManager:
    """
    Full lifecycle manager for AMM liquidity positions.
    Supports concentrated Uniswap V3-style positions with
    auto-rebalancing and impermanent loss monitoring.

    API key starts with pf_live_ prefix for Purple Flea services.
    """

    TICK_BASE = 1.0001
    MIN_COLLECT_INTERVAL = 3600   # 1 hour minimum between fee collections
    DEFAULT_REBALANCE_BUFFER = 0.02  # 2% buffer before triggering rebalance

    def __init__(
        self,
        api_key: str,
        wallet_id: str,
        slippage_tolerance: float = 0.005,  # 0.5%
        max_gas_gwei: int = 30,
        rebalance_buffer: float = 0.02,
    ):
        self.api_key = api_key
        self.wallet_id = wallet_id
        self.slippage = slippage_tolerance
        self.max_gas = max_gas_gwei
        self.rebalance_buffer = rebalance_buffer
        self.positions: Dict[str, LPPosition] = {}
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "X-Wallet-ID": wallet_id,
        })

    # ──────────────────────────────────────────────────────────────
    # TICK MATH UTILITIES
    # ──────────────────────────────────────────────────────────────

    def price_to_tick(self, price: float) -> int:
        """Convert a price to the nearest valid tick."""
        return math.floor(math.log(price) / math.log(self.TICK_BASE))

    def tick_to_price(self, tick: int) -> float:
        """Convert a tick index back to price."""
        return self.TICK_BASE ** tick

    def sqrt_price(self, price: float) -> float:
        return math.sqrt(price)

    def compute_range_multiplier(self, p_lower: float, p_upper: float) -> float:
        """
        Capital efficiency multiplier for a concentrated range.
        Tight ranges have higher multipliers but require more rebalancing.
        """
        ratio = math.sqrt(p_upper / p_lower)
        return ratio / (ratio - 1)

    def compute_liquidity_amounts(
        self,
        current_price: float,
        p_lower: float,
        p_upper: float,
        capital_usd: float,
    ) -> Tuple[float, float]:
        """
        Given a capital budget and price range, compute the
        token0 and token1 amounts to deposit (50/50 split by value).
        """
        sq_current = self.sqrt_price(current_price)
        sq_lower   = self.sqrt_price(p_lower)
        sq_upper   = self.sqrt_price(p_upper)

        # Amount of token0 needed per unit of liquidity
        delta0 = (1 / sq_current - 1 / sq_upper)
        # Amount of token1 needed per unit of liquidity
        delta1 = sq_current - sq_lower

        # Token0 value in USD (price = token1/token0)
        value0_per_unit = delta0 * current_price
        value1_per_unit = delta1

        total_per_unit = value0_per_unit + value1_per_unit
        units = capital_usd / total_per_unit

        amount0 = units * delta0
        amount1 = units * delta1
        return amount0, amount1

    # ──────────────────────────────────────────────────────────────
    # CORE POSITION MANAGEMENT
    # ──────────────────────────────────────────────────────────────

    def add_liquidity(
        self,
        pool_address: str,
        token0: str,
        token1: str,
        fee_tier: float,
        capital_usd: float,
        range_pct: float = 0.10,   # ±10% around current price
        current_price: Optional[float] = None,
    ) -> LPPosition:
        """
        Open a new concentrated liquidity position.

        Args:
            pool_address: DEX pool contract address
            token0, token1: Token symbols (e.g. 'USDC', 'ETH')
            fee_tier: Pool fee tier (0.0005, 0.003, or 0.01)
            capital_usd: Total capital to deploy in USD
            range_pct: Half-width of price range as a fraction
            current_price: Override current price (fetched if None)

        Returns:
            LPPosition tracking object
        """
        if current_price is None:
            current_price = self._fetch_pool_price(pool_address)

        p_lower = current_price * (1 - range_pct)
        p_upper = current_price * (1 + range_pct)

        tick_lower = self.price_to_tick(p_lower)
        tick_upper = self.price_to_tick(p_upper)

        amount0, amount1 = self.compute_liquidity_amounts(
            current_price, p_lower, p_upper, capital_usd
        )

        multiplier = self.compute_range_multiplier(p_lower, p_upper)
        logger.info(
            f"Opening LP: {token0}/{token1} range [{p_lower:.4f}, {p_upper:.4f}] "
            f"multiplier={multiplier:.1f}x capital=${capital_usd:,.0f}"
        )

        payload = {
            "action": "add_liquidity",
            "pool_address": pool_address,
            "tick_lower": tick_lower,
            "tick_upper": tick_upper,
            "amount0_desired": amount0,
            "amount1_desired": amount1,
            "amount0_min": amount0 * (1 - self.slippage),
            "amount1_min": amount1 * (1 - self.slippage),
            "max_gas_gwei": self.max_gas,
        }

        resp = self.session.post(f"{PURPLE_FLEA_BASE}/trading/liquidity", json=payload)
        resp.raise_for_status()
        data = resp.json()

        pos = LPPosition(
            position_id=data["position_id"],
            pool_address=pool_address,
            token0=token0,
            token1=token1,
            fee_tier=fee_tier,
            tick_lower=tick_lower,
            tick_upper=tick_upper,
            liquidity=data["liquidity"],
            entry_price=current_price,
            token0_deposited=amount0,
            token1_deposited=amount1,
        )
        self.positions[pos.position_id] = pos
        logger.info(f"Position opened: {pos.position_id}")
        return pos

    def remove_liquidity(
        self,
        position_id: str,
        pct: float = 1.0,   # 1.0 = 100% withdrawal
        collect_fees: bool = True,
    ) -> Dict:
        """
        Remove some or all liquidity from a position.
        Optionally collect any unclaimed fees first.
        """
        pos = self.positions[position_id]

        if collect_fees:
            self.collect_fees(position_id)

        liquidity_to_remove = int(pos.liquidity * pct)
        payload = {
            "action": "remove_liquidity",
            "position_id": position_id,
            "liquidity": liquidity_to_remove,
            "amount0_min": 0,
            "amount1_min": 0,
            "max_gas_gwei": self.max_gas,
        }

        resp = self.session.post(f"{PURPLE_FLEA_BASE}/trading/liquidity", json=payload)
        resp.raise_for_status()
        result = resp.json()

        logger.info(
            f"Removed {pct*100:.0f}% of {position_id}: "
            f"got {result.get('amount0_returned', 0):.4f} {pos.token0} + "
            f"{result.get('amount1_returned', 0):.4f} {pos.token1}"
        )

        if pct >= 1.0:
            del self.positions[position_id]

        return result

    def collect_fees(self, position_id: str) -> Dict[str, float]:
        """
        Claim accrued trading fees for a position.
        Returns {'fee0': float, 'fee1': float, 'fee_usd': float}.
        """
        pos = self.positions[position_id]
        now = time.time()

        if now - pos.last_collected < self.MIN_COLLECT_INTERVAL:
            logger.debug(f"Skipping fee collection for {position_id} — too recent")
            return {"fee0": 0, "fee1": 0, "fee_usd": 0}

        payload = {
            "action": "collect_fees",
            "position_id": position_id,
        }

        resp = self.session.post(f"{PURPLE_FLEA_BASE}/trading/liquidity/collect", json=payload)
        resp.raise_for_status()
        data = resp.json()

        fee0 = data.get("fee0", 0)
        fee1 = data.get("fee1", 0)
        fee_usd = data.get("fee_usd", 0)

        pos.fees_collected_token0 += fee0
        pos.fees_collected_token1 += fee1
        pos.last_collected = now

        logger.info(
            f"Fees collected for {position_id}: "
            f"{fee0:.6f} {pos.token0} + {fee1:.6f} {pos.token1} "
            f"(${fee_usd:.2f})"
        )
        return {"fee0": fee0, "fee1": fee1, "fee_usd": fee_usd}

    def rebalance_range(
        self,
        position_id: str,
        new_range_pct: float = 0.10,
        current_price: Optional[float] = None,
    ) -> LPPosition:
        """
        Close existing position and reopen centered on current price.
        Called when price has moved outside the active range.
        """
        pos = self.positions[position_id]
        logger.info(f"Rebalancing {position_id} (rebalance #{pos.rebalance_count + 1})")

        if current_price is None:
            current_price = self._fetch_pool_price(pos.pool_address)

        # Step 1: Remove all liquidity + collect fees
        removed = self.remove_liquidity(position_id, pct=1.0, collect_fees=True)

        # Step 2: Calculate new capital (returned amounts converted to USD)
        new_capital_usd = removed.get("total_usd", 0)
        if new_capital_usd == 0:
            new_capital_usd = (
                removed.get("amount0_returned", 0) * current_price +
                removed.get("amount1_returned", 0)
            )

        # Step 3: Open new centered position
        new_pos = self.add_liquidity(
            pool_address=pos.pool_address,
            token0=pos.token0,
            token1=pos.token1,
            fee_tier=pos.fee_tier,
            capital_usd=new_capital_usd,
            range_pct=new_range_pct,
            current_price=current_price,
        )
        new_pos.rebalance_count = pos.rebalance_count + 1
        new_pos.fees_collected_token0 = pos.fees_collected_token0
        new_pos.fees_collected_token1 = pos.fees_collected_token1
        return new_pos

    # ──────────────────────────────────────────────────────────────
    # IMPERMANENT LOSS
    # ──────────────────────────────────────────────────────────────

    def compute_impermanent_loss(
        self,
        position_id: str,
        current_price: Optional[float] = None,
    ) -> float:
        """
        Compute IL as a fraction of hold value.
        Returns a negative number (loss) or 0 if no IL.

        Formula: IL = 2*sqrt(k) / (1+k) - 1
        where k = current_price / entry_price
        """
        pos = self.positions[position_id]
        if current_price is None:
            current_price = self._fetch_pool_price(pos.pool_address)

        k = current_price / pos.entry_price
        il = (2 * math.sqrt(k) / (1 + k)) - 1
        return il   # always <= 0

    def is_in_range(self, position_id: str, current_price: float) -> bool:
        """Return True if current price falls within the position's tick range."""
        pos = self.positions[position_id]
        current_tick = self.price_to_tick(current_price)
        return pos.tick_lower <= current_tick <= pos.tick_upper

    def distance_to_boundary(self, position_id: str, current_price: float) -> float:
        """
        Returns the fractional distance from current price to
        the nearest range boundary. Negative = out of range.
        """
        pos = self.positions[position_id]
        p_lower = self.tick_to_price(pos.tick_lower)
        p_upper = self.tick_to_price(pos.tick_upper)
        dist_up   = (p_upper - current_price) / current_price
        dist_down = (current_price - p_lower) / current_price
        return min(dist_up, dist_down)

    # ──────────────────────────────────────────────────────────────
    # MONITORING LOOP
    # ──────────────────────────────────────────────────────────────

    def monitor_positions(self, poll_interval: int = 60):
        """
        Continuous monitoring loop. At each interval:
        1. Fetch current price for each position's pool
        2. Check if in-range; rebalance if out-of-range
        3. Collect fees if accrual threshold exceeded
        4. Log IL and cumulative P&L

        Runs forever — deploy with PM2 or similar.
        """
        logger.info(f"Starting monitor loop ({len(self.positions)} positions, poll={poll_interval}s)")

        while True:
            for pid, pos in list(self.positions.items()):
                try:
                    price = self._fetch_pool_price(pos.pool_address)
                    in_range = self.is_in_range(pid, price)
                    il = self.compute_impermanent_loss(pid, price)
                    dist = self.distance_to_boundary(pid, price)

                    logger.info(
                        f"[{pid[:8]}] {pos.token0}/{pos.token1} "
                        f"price={price:.4f} in_range={in_range} "
                        f"IL={il*100:.2f}% boundary_dist={dist*100:.1f}%"
                    )

                    if not in_range or dist < self.rebalance_buffer:
                        logger.warning(f"[{pid[:8]}] Out of range — triggering rebalance")
                        self.rebalance_range(pid, current_price=price)

                    self.collect_fees(pid)

                except Exception as e:
                    logger.error(f"Error monitoring {pid}: {e}")

            time.sleep(poll_interval)

    # ──────────────────────────────────────────────────────────────
    # HELPERS
    # ──────────────────────────────────────────────────────────────

    def _fetch_pool_price(self, pool_address: str) -> float:
        """Fetch current spot price from Purple Flea Trading API."""
        resp = self.session.get(
            f"{PURPLE_FLEA_BASE}/trading/pools/{pool_address}/price"
        )
        resp.raise_for_status()
        return resp.json()["price"]

    def position_summary(self, position_id: str) -> Dict:
        """Return a full P&L summary for a position."""
        pos = self.positions[position_id]
        price = self._fetch_pool_price(pos.pool_address)
        il = self.compute_impermanent_loss(position_id, price)
        in_range = self.is_in_range(position_id, price)

        return {
            "position_id": position_id,
            "pair": f"{pos.token0}/{pos.token1}",
            "fee_tier": pos.fee_tier,
            "entry_price": pos.entry_price,
            "current_price": price,
            "in_range": in_range,
            "impermanent_loss_pct": il * 100,
            "fees_collected_token0": pos.fees_collected_token0,
            "fees_collected_token1": pos.fees_collected_token1,
            "rebalance_count": pos.rebalance_count,
            "age_hours": (time.time() - pos.created_at) / 3600,
        }

Impermanent Loss: Monitoring and Mitigation

Impermanent loss (IL) is the opportunity cost of providing liquidity compared to simply holding both tokens. It is "impermanent" because if the price returns to the entry level, IL disappears. In practice, many LPs realize IL when they withdraw, making it permanent.

IL Formula Breakdown

Mitigation Strategy 1: Hedge with Perpetuals

An agent can hedge IL by opening an opposing perpetual futures position. If you provide ETH/USDC liquidity, you are effectively long ETH when price rises (position shifts toward USDC) and short ETH when price falls (position shifts toward ETH). The perpetual hedge cancels this directional exposure.

def compute_lp_delta(
    current_price: float,
    p_lower: float,
    p_upper: float,
    liquidity: float,
) -> float:
    """
    Compute the net token0 delta of a concentrated LP position.
    Returns tokens of token0 (negative = net short token0 equivalent).
    Derivative of V3 position value w.r.t. price.
    """
    sq_p     = math.sqrt(current_price)
    sq_lower = math.sqrt(p_lower)
    sq_upper = math.sqrt(p_upper)

    if current_price <= p_lower:
        # Fully in token0, delta = total token0 amount
        return liquidity * (1/sq_lower - 1/sq_upper)
    elif current_price >= p_upper:
        # Fully in token1, delta = 0 (no more token0)
        return 0
    else:
        # Partial: in-range delta
        return liquidity * (1/sq_p - 1/sq_upper)


def place_perp_hedge(
    lm: LiquidityManager,
    position_id: str,
    current_price: float,
) -> Dict:
    """
    Open a perpetual short to hedge LP delta.
    Automatically sizes the position to neutralize directional exposure.
    """
    pos = lm.positions[position_id]
    p_lower = lm.tick_to_price(pos.tick_lower)
    p_upper = lm.tick_to_price(pos.tick_upper)

    delta = compute_lp_delta(current_price, p_lower, p_upper, pos.liquidity)
    hedge_size = abs(delta)   # Short this many units of token0

    if hedge_size < 0.001:
        return {"status": "no_hedge_needed"}

    payload = {
        "action": "open_perp",
        "symbol": f"{pos.token0}/USDC-PERP",
        "side": "short",
        "size": hedge_size,
        "order_type": "market",
        "reduce_only": False,
    }

    resp = lm.session.post(f"{PURPLE_FLEA_BASE}/trading/perps/order", json=payload)
    resp.raise_for_status()
    logger.info(f"Perp hedge placed: short {hedge_size:.4f} {pos.token0}")
    return resp.json()

Mitigation Strategy 2: Tight Ranges on Stable Pairs

For correlated assets (USDC/USDT, stETH/ETH, BTC/WBTC), IL is negligible because the price ratio rarely moves far. An agent can deploy capital in an extremely tight ±0.1% range and capture enormous fee income with minimal IL risk. These positions still need monitoring in case of depegs.

Mitigation Strategy 3: Range Expansion on Volatility Alerts

When an agent detects a volatility spike (e.g., implied vol from options markets exceeds 2 sigma), it can proactively widen the range before price escapes, sacrificing some capital efficiency to avoid the cost of an emergency rebalance. This is a classic volatility-adaptive LP strategy.

Auto-Rebalancing Strategy

Rebalancing closes the current position and reopens it centered on the new price. Each rebalance incurs: (1) gas costs, (2) swap fees to re-balance token ratios, and (3) slippage. The key decision: when does the cost of rebalancing exceed the cost of staying out-of-range?

Rebalance Trigger Logic

from dataclasses import dataclass

@dataclass
class RebalanceDecision:
    should_rebalance: bool
    reason: str
    estimated_gas_usd: float
    opportunity_cost_usd: float
    net_benefit_usd: float


def should_rebalance(
    lm: LiquidityManager,
    position_id: str,
    current_price: float,
    pool_daily_volume_usd: float,
    gas_price_gwei: int = 20,
) -> RebalanceDecision:
    """
    Decide whether rebalancing a position is economically worthwhile.

    Compares:
    - Cost of rebalancing (gas + slippage)
    - Cost of staying out-of-range (zero fees earned)
    """
    pos = lm.positions[position_id]

    # If still in range, no action needed
    if lm.is_in_range(position_id, current_price):
        return RebalanceDecision(
            should_rebalance=False,
            reason="in_range",
            estimated_gas_usd=0,
            opportunity_cost_usd=0,
            net_benefit_usd=0,
        )

    # Estimate gas cost of rebalance (2 txns: burn + mint)
    gas_units = 300_000  # typical V3 position mgmt gas
    eth_price = 3000    # fetch from oracle in production
    gas_cost_usd = (gas_price_gwei * gas_units * 1e-9) * eth_price

    # Slippage cost (0.1% of position value to re-balance ratios)
    position_value_usd = (
        pos.token0_deposited * current_price + pos.token1_deposited
    )
    slippage_cost_usd = position_value_usd * 0.001
    total_rebalance_cost = gas_cost_usd + slippage_cost_usd

    # Opportunity cost: fees missed per hour out-of-range
    # Assume our share = (position_value / total_tvl), proxy 1% of volume
    daily_fee_income = pool_daily_volume_usd * pos.fee_tier * 0.001
    hourly_fee_income = daily_fee_income / 24

    # Hours to recoup rebalance cost from fee income
    breakeven_hours = total_rebalance_cost / hourly_fee_income if hourly_fee_income > 0 else float('inf')

    # Always rebalance if out-of-range AND breakeven < 48h
    net_benefit = hourly_fee_income * 24 - total_rebalance_cost

    return RebalanceDecision(
        should_rebalance=breakeven_hours < 48,
        reason=f"breakeven_in_{breakeven_hours:.1f}h",
        estimated_gas_usd=total_rebalance_cost,
        opportunity_cost_usd=hourly_fee_income * 24,
        net_benefit_usd=net_benefit,
    )

Tracking LP Income with the Wallet API

LP income — fees from swap activity — is distinct from trading P&L and should be tracked separately for accurate performance attribution. The Purple Flea Wallet API supports labeled income streams so your agent can report fee income independently from perpetual hedge P&L.

WALLET_API = "https://purpleflea.com/api/v1/wallet"

def record_fee_income(
    session: requests.Session,
    wallet_id: str,
    position_id: str,
    fee_usd: float,
    token0_amount: float,
    token1_amount: float,
    token0: str,
    token1: str,
) -> Dict:
    """
    Log fee collection event to Purple Flea Wallet API.
    Tags the transaction as 'lp_fee_income' for reporting.
    """
    payload = {
        "wallet_id": wallet_id,
        "event_type": "lp_fee_income",
        "amount_usd": fee_usd,
        "metadata": {
            "position_id": position_id,
            "token0": token0,
            "token0_amount": token0_amount,
            "token1": token1,
            "token1_amount": token1_amount,
            "source": "amm_liquidity",
        },
        "timestamp": time.time(),
    }
    resp = session.post(f"{WALLET_API}/transactions", json=payload)
    resp.raise_for_status()
    return resp.json()


def get_lp_income_summary(
    session: requests.Session,
    wallet_id: str,
    since_timestamp: float,
) -> Dict:
    """Fetch total LP fee income since a given timestamp."""
    resp = session.get(
        f"{WALLET_API}/transactions",
        params={
            "wallet_id": wallet_id,
            "event_type": "lp_fee_income",
            "since": since_timestamp,
        }
    )
    resp.raise_for_status()
    txns = resp.json()["transactions"]

    total_usd = sum(t["amount_usd"] for t in txns)
    return {
        "total_fee_income_usd": total_usd,
        "event_count": len(txns),
        "avg_per_event": total_usd / len(txns) if txns else 0,
    }

Full Agent Example

Putting it all together — an agent that opens a concentrated ETH/USDC position, monitors it continuously, hedges IL with perps, and reports income to the Wallet API. Ready to deploy with PM2 on any VPS.

import os
import time
import threading

# Purple Flea credentials — use pf_live_ prefix for production keys
API_KEY   = os.environ["PURPLE_FLEA_API_KEY"]    # pf_live_...
WALLET_ID = os.environ["PURPLE_FLEA_WALLET_ID"]

ETH_USDC_POOL = "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"

def main():
    lm = LiquidityManager(
        api_key=API_KEY,
        wallet_id=WALLET_ID,
        slippage_tolerance=0.003,
        max_gas_gwei=25,
        rebalance_buffer=0.03,   # rebalance when within 3% of boundary
    )

    # Deploy $10,000 into ETH/USDC 0.3% pool, ±8% range
    pos = lm.add_liquidity(
        pool_address=ETH_USDC_POOL,
        token0="ETH",
        token1="USDC",
        fee_tier=0.003,
        capital_usd=10_000,
        range_pct=0.08,
    )
    logger.info(f"Position opened: {pos.position_id}")

    # Place initial delta hedge via perps
    current_price = lm._fetch_pool_price(ETH_USDC_POOL)
    hedge = place_perp_hedge(lm, pos.position_id, current_price)
    logger.info(f"Hedge placed: {hedge}")

    # Start monitoring in main thread (blocks forever)
    lm.monitor_positions(poll_interval=60)


if __name__ == "__main__":
    main()

Choosing Your LP Strategy

Not all liquidity provision strategies suit every agent. The right approach depends on available capital, compute budget, risk tolerance, and the pairs you are liquidity-providing for.

LP Income with Purple Flea

Purple Flea provides the financial infrastructure to support the complete liquidity provision lifecycle for AI agents. From initial capital via the Faucet, to hedging via the Trading API, to income separation via the Wallet API, and agent-to-agent payments for multi-agent coordination via the Escrow service.

Conclusion

Concentrated liquidity provision is one of the highest-return passive income strategies available to AI agents with access to on-chain capital. The mathematics are well-understood, the risks are manageable with proper hedging, and the entire workflow — from capital deployment to fee collection to IL hedging — can be fully automated.

What gives AI agents an edge over human LPs is not strategy selection but execution frequency. A human LP rebalances weekly; an agent rebalances within minutes of price leaving the range. This compounding advantage means agents systematically capture more fees, suffer less idle capital, and accumulate positions that outperform their human counterparts over any meaningful time horizon.

Start with the Purple Flea Faucet for free capital, connect to the Trading API for LP pool access and perpetual hedges, and use the Wallet API to track your income streams cleanly. The infrastructure is ready — your agent just needs to run.