Market Making for AI Agents: Liquidity Provision on Purple Flea

Market making is one of the most enduring profitable strategies in financial markets — and AI agents are uniquely well-suited to it. Where human traders fatigue, miss quote updates, or hesitate during volatility, agents run deterministic quote engines 24/7. This guide covers everything an agent needs to act as a liquidity provider on Purple Flea Trading: bid-ask spread theory, order book positioning, inventory risk management, grid strategies, and funding rate arbitrage.

What Is Market Making?

A market maker simultaneously posts a bid (the price it will buy at) and an ask (the price it will sell at). The difference between these prices is the spread. Every time a taker fills against one of your quotes, you earn half the spread minus fees. Over thousands of fills per day, this compounds into a consistent income stream.

On perpetual futures markets like Purple Flea Trading, market makers also interact with the funding rate mechanism. When longs outnumber shorts, longs pay shorts a periodic funding payment. A market maker with a balanced book earns this funding as pure yield on top of spread income.

Bid-Ask Spread Mechanics

The optimal spread balances two competing forces:

• Adverse selection risk — the risk that a taker filling your quote knows something you don't (e.g. has inside information about price direction). Wider spreads protect against this.
• Fill rate — if your spread is too wide, takers go to other liquidity providers. You earn zero from unfilled quotes. Tighter spreads attract more fills.

The classic Avellaneda-Stoikov model sets the reservation price (the midpoint you quote around) and the half-spread as follows:

# Avellaneda-Stoikov market making model
import math

def optimal_spread(
    sigma: float,       # realized volatility (hourly)
    gamma: float,       # risk aversion parameter (0.01 to 0.1)
    T: float,           # time horizon in hours (e.g. 1.0)
    t: float,           # elapsed time in hours
    k: float,           # order book depth parameter (typically 1.5)
) -> tuple[float, float]:
    """
    Returns (half_spread, optimal_delta) for the current market conditions.
    half_spread: how far to post bids and asks from mid
    """
    # Variance term
    variance = sigma ** 2

    # Reservation spread (core of A-S model)
    reservation_spread = gamma * variance * (T - t) + (2 / gamma) * math.log(1 + gamma / k)

    half_spread = reservation_spread / 2
    return half_spread, reservation_spread

# Example: BTC-PERP, 2% hourly vol, moderate risk aversion
mid_price = 65000.0
half_spread, spread = optimal_spread(sigma=0.02, gamma=0.05, T=1.0, t=0.5, k=1.5)
bid = mid_price * (1 - half_spread)
ask = mid_price * (1 + half_spread)
print(f"Bid: {bid:.2f}  Ask: {ask:.2f}  Spread: {spread*100:.3f}%")

Order Book Management

A production market maker doesn't place single orders — it maintains a ladder of quotes at multiple price levels. This spreads risk across the order book and captures fills at different price points:

class OrderLadder:
    """Maintains a multi-level quote ladder around the current mid price."""

    def __init__(self, n_levels: int = 5, level_spacing_bps: float = 10):
        self.n_levels = n_levels
        self.level_spacing = level_spacing_bps / 10000  # convert bps to decimal
        self.active_orders: dict[str, dict] = {}

    def compute_quotes(self, mid: float, base_half_spread: float, base_size: float) -> list[dict]:
        """Generate a list of bid/ask quotes for all ladder levels."""
        quotes = []
        for i in range(self.n_levels):
            # Each level is further from mid by level_spacing
            offset = base_half_spread + (i * self.level_spacing)
            # Size increases with distance (skewed toward edge)
            size = base_size * (1 + i * 0.5)

            quotes.append({
                "side": "bid",
                "price": mid * (1 - offset),
                "size": size,
                "level": i
            })
            quotes.append({
                "side": "ask",
                "price": mid * (1 + offset),
                "size": size,
                "level": i
            })
        return quotes

    def should_requote(self, current_mid: float, quoted_mid: float, threshold_bps: float = 5) -> bool:
        """Returns True if mid has moved enough to warrant canceling and requoting."""
        drift_bps = abs(current_mid - quoted_mid) / quoted_mid * 10000
        return drift_bps > threshold_bps

Inventory Risk Management

The greatest danger for a market maker is inventory accumulation. If price moves against you while you're accumulating a position, you can lose far more than you've earned from spreads. Robust inventory management is essential:

class InventoryManager:
    """Tracks net position and skews quotes to manage inventory risk."""

    def __init__(self, max_inventory_usdc: float = 500):
        self.max_inventory = max_inventory_usdc
        self.net_position_usdc = 0.0   # positive = long, negative = short
        self.fills_today = []

    def inventory_ratio(self) -> float:
        """Returns -1 to +1, where +1 = fully long (at max), -1 = fully short."""
        return self.net_position_usdc / self.max_inventory

    def skewed_quotes(self, mid: float, base_half_spread: float) -> tuple[float, float]:
        """
        Returns (bid_price, ask_price) skewed to reduce inventory.
        When long, ask more aggressively (lower ask) and bid less aggressively.
        When short, bid more aggressively (higher bid) and ask less aggressively.
        """
        ratio = self.inventory_ratio()
        skew = ratio * base_half_spread * 0.5   # max 50% skew

        bid = mid * (1 - base_half_spread) - mid * skew
        ask = mid * (1 + base_half_spread) - mid * skew

        return bid, ask

    def should_pause_quoting(self) -> bool:
        """Stop quoting if inventory is at 90% of limit."""
        return abs(self.inventory_ratio()) > 0.9

    def emergency_reduce(self, client) -> dict:
        """Place a market order to cut inventory in half immediately."""
        reduce_size = abs(self.net_position_usdc) * 0.5
        side = "short" if self.net_position_usdc > 0 else "long"
        return client.open_position(
            market=self.market,
            side=side,
            size_usdc=reduce_size,
            order_type="market"
        )

Grid Strategy as Market Making

A grid strategy is the simplest form of market making: place buy orders at fixed intervals below mid price and sell orders at fixed intervals above. When a buy fills, immediately place a sell at buy_price + grid_spacing, and vice versa. Profit accumulates from each round-trip.

import asyncio, httpx, os
from dataclasses import dataclass, field

BASE_URL = "https://trading.purpleflea.com"
HEADERS = {"Authorization": f"Bearer {os.getenv('PURPLE_FLEA_KEY')}"}

@dataclass
class GridConfig:
    market: str = "BTC-PERP"
    lower_price: float = 62000
    upper_price: float = 68000
    grid_count: int = 10
    order_size_usdc: float = 20

class GridMarketMaker:
    def __init__(self, config: GridConfig):
        self.config = config
        self.grid_spacing = (config.upper_price - config.lower_price) / config.grid_count
        self.active_orders: dict[float, str] = {}  # price -> order_id
        self.pnl_usdc = 0.0
        self.fills = 0

    def grid_levels(self) -> list[float]:
        """Generate all price levels for the grid."""
        levels = []
        price = self.config.lower_price
        while price <= self.config.upper_price:
            levels.append(round(price, 2))
            price += self.grid_spacing
        return levels

    async def place_grid(self, current_price: float):
        """Place all grid orders — bids below current price, asks above."""
        async with httpx.AsyncClient() as client:
            for level in self.grid_levels():
                side = "long" if level < current_price else "short"
                r = await client.post(
                    f"{BASE_URL}/api/orders",
                    json={
                        "market": self.config.market,
                        "side": side,
                        "size": self.config.order_size_usdc,
                        "type": "limit",
                        "price": level
                    },
                    headers=HEADERS
                )
                if r.status_code == 200:
                    order = r.json()
                    self.active_orders[level] = order["order_id"]
        print(f"Grid placed: {len(self.active_orders)} orders")

    async def on_fill(self, fill_price: float, fill_side: str):
        """Handle a fill: place counter-order at fill_price ± grid_spacing."""
        self.fills += 1
        # Place the counter-order on the other side
        if fill_side == "long":
            counter_price = fill_price + self.grid_spacing
            counter_side = "short"
        else:
            counter_price = fill_price - self.grid_spacing
            counter_side = "long"

        # Profit from each round-trip = grid_spacing - fees
        round_trip_pnl = self.grid_spacing / fill_price * self.config.order_size_usdc
        self.pnl_usdc += round_trip_pnl
        print(f"Fill #{self.fills}: {fill_side} @ {fill_price:.2f} | PnL: +${round_trip_pnl:.4f} | Total: ${self.pnl_usdc:.4f}")

Adapting to Volatility

Static grids break during high volatility — price can blow through multiple grid levels in seconds, causing rapid inventory accumulation. Volatility-adaptive market makers widen spreads and reduce size when realized vol spikes:

class VolatilityAdapter:
    def __init__(self, window: int = 20):
        self.window = window
        self.prices: list[float] = []

    def update(self, price: float):
        self.prices.append(price)
        if len(self.prices) > self.window:
            self.prices.pop(0)

    def realized_vol(self) -> float:
        """Annualized realized volatility from recent prices."""
        if len(self.prices) < 2:
            return 0.5  # default 50% annualized
        import statistics, math
        returns = [math.log(self.prices[i] / self.prices[i-1])
                   for i in range(1, len(self.prices))]
        std = statistics.stdev(returns)
        # Scale to annualized (assuming 1-min ticks, 525600 per year)
        return std * math.sqrt(525600)

    def vol_regime(self) -> str:
        vol = self.realized_vol()
        if vol < 0.4:   return "low"
        elif vol < 0.8: return "medium"
        else:           return "high"

    def spread_multiplier(self) -> float:
        regime = self.vol_regime()
        return {"low": 1.0, "medium": 1.5, "high": 3.0}[regime]

    def size_multiplier(self) -> float:
        regime = self.vol_regime()
        return {"low": 1.0, "medium": 0.7, "high": 0.3}[regime]

Funding Rate Arbitrage

Perpetual futures use a funding rate mechanism to keep the futures price anchored to the spot price. Every 8 hours, longs pay shorts (or vice versa) based on the funding rate. A market maker with a near-neutral book position can tilt slightly toward the profitable side and collect funding as additional income:

async def check_funding_opportunity(market: str, min_rate_bps: float = 3.0) -> dict:
    """
    Check if funding rate is high enough to warrant a deliberate tilt.
    Returns recommendation: 'lean_long', 'lean_short', or 'neutral'.
    """
    async with httpx.AsyncClient() as client:
        r = await client.get(
            f"{BASE_URL}/api/ticker/{market}",
            headers=HEADERS
        )
        data = r.json()

    funding_rate_8h = data["funding_rate"]       # e.g. 0.0003 = 0.03% per 8h
    funding_rate_bps = funding_rate_8h * 10000    # convert to basis points

    if abs(funding_rate_bps) < min_rate_bps:
        return {"action": "neutral", "rate_bps": funding_rate_bps}

    # Positive funding: longs pay shorts → lean short
    # Negative funding: shorts pay longs → lean long
    action = "lean_short" if funding_rate_8h > 0 else "lean_long"
    annual_yield = abs(funding_rate_8h) * 3 * 365 * 100  # 3 payments/day * 365
    return {
        "action": action,
        "rate_bps": funding_rate_bps,
        "annualized_yield_pct": annual_yield,
        "next_funding_at": data["next_funding_at"]
    }

Complete MarketMaker Class

Here is a full, production-ready MarketMaker class that integrates spread optimization, inventory management, volatility adaptation, and funding rate awareness:

class MarketMaker:
    """
    Full market making agent for Purple Flea Trading.
    Integrates Avellaneda-Stoikov quoting, inventory skew,
    volatility regime adaptation, and funding rate tilting.
    """

    def __init__(
        self,
        market: str,
        base_half_spread_bps: float = 8,
        max_inventory_usdc: float = 1000,
        quote_size_usdc: float = 50
    ):
        self.market = market
        self.base_half_spread = base_half_spread_bps / 10000
        self.inv_mgr = InventoryManager(max_inventory_usdc)
        self.vol_adapter = VolatilityAdapter(window=20)
        self.quote_size = quote_size_usdc
        self.running = False
        self.stats = {"fills": 0, "pnl": 0.0, "uptime_s": 0}

    async def get_mid_price(self) -> float:
        async with httpx.AsyncClient() as client:
            r = await client.get(f"{BASE_URL}/api/ticker/{self.market}", headers=HEADERS)
            return r.json()["price"]

    async def place_quotes(self, mid: float):
        # Apply volatility and inventory adjustments
        adjusted_spread = self.base_half_spread * self.vol_adapter.spread_multiplier()
        adjusted_size = self.quote_size * self.vol_adapter.size_multiplier()
        bid, ask = self.inv_mgr.skewed_quotes(mid, adjusted_spread)

        if self.inv_mgr.should_pause_quoting():
            print("Inventory limit reached — pausing quotes")
            return

        async with httpx.AsyncClient() as client:
            # Place bid and ask simultaneously
            await asyncio.gather(
                client.post(f"{BASE_URL}/api/orders", headers=HEADERS,
                    json={"market": self.market, "side": "long",
                          "size": adjusted_size, "type": "limit", "price": bid}),
                client.post(f"{BASE_URL}/api/orders", headers=HEADERS,
                    json={"market": self.market, "side": "short",
                          "size": adjusted_size, "type": "limit", "price": ask})
            )

    async def run(self, quote_interval_s: float = 5.0):
        """Main market making loop."""
        self.running = True
        print(f"MarketMaker started on {self.market}")

        while self.running:
            try:
                mid = await self.get_mid_price()
                self.vol_adapter.update(mid)
                await self.place_quotes(mid)
                await asyncio.sleep(quote_interval_s)
            except Exception as e:
                print(f"Error in MM loop: {e}")
                await asyncio.sleep(10)  # back off on errors

# Run it
async def main():
    mm = MarketMaker(market="BTC-PERP", base_half_spread_bps=8, max_inventory_usdc=500)
    await mm.run(quote_interval_s=3.0)

asyncio.run(main())

Performance Metrics

---

Market making is a long-horizon strategy — don't evaluate it on a single day's P&L. Backtest thoroughly, start with small position sizes, monitor inventory ratios obsessively, and let the law of large numbers work in your favor. The edge is real; the risk is inventory accumulation during trending markets. Manage that, and market making on Purple Flea Trading can be a durable income source for your agent.