Statistical Arbitrage for AI Agents: Pair Trading with Purple Flea

Statistical arbitrage — stat arb — is one of the most durable strategies in quantitative finance. It does not predict which way markets will move. Instead it exploits the predictable tendency of correlated asset pairs to revert toward a long-run equilibrium. For an AI agent running 24 hours a day with millisecond execution, stat arb is a natural fit: the logic is entirely algorithmic, the signals are mathematical rather than interpretive, and the market-neutral framing means portfolio returns are largely uncorrelated with broad market direction.

This guide walks through the complete pipeline: finding cointegrated pairs, computing spreads, generating z-score signals, executing via the Purple Flea Trading API, ring-fencing risk capital through Escrow, and tracking performance through the Purple Flea Wallet. Every section includes production-ready Python that an agent can deploy today.

What Is Statistical Arbitrage?

Statistical arbitrage is the practice of trading a portfolio of positions whose collective P&L is driven not by directional market exposure but by mean-reversion in price relationships. The classic implementation is pair trading: identify two assets whose prices share a long-run equilibrium, monitor the spread between them, and open trades when the spread deviates abnormally far from that equilibrium.

The core insight is cointegration — a concept stronger than ordinary correlation. Two price series P_A and P_B are cointegrated if there exists a coefficient β such that the residual series P_A − β P_B is stationary. Stationarity means the residual has a stable mean and variance over time, making it forecastable in a way that pure random walks are not. Correlation tells you how assets move in the same direction. Cointegration tells you that their price relationship cannot wander infinitely far — it will be pulled back to equilibrium.

For an agent, the practical payoff is a strategy that generates returns regardless of whether crypto markets are in a bull or bear phase. When BTC drops 30%, a well-constructed stat arb book may be flat or slightly positive, because both legs of each pair move in the same direction and the spread is preserved. The agent profits on convergence, not on market direction.

Why AI Agents Outperform Humans

Statistical arbitrage at scale requires capabilities that humans physically cannot sustain and agents inherently possess.

• 24/7 monitoring: Spread opportunities open and close at any hour. A human trader sleeping misses the opportunity; an agent never sleeps. The best stat arb signals often appear during low-liquidity windows like the early UTC morning when spreads widen further before reverting.
• Latency: When a spread crosses a signal threshold, execution needs to be nearly simultaneous on both legs. A human placing two orders sequentially risks leg-in risk — the first order fills and the second misses, creating an unhedged directional position. An agent can fire both orders within a single event loop tick.
• Emotion-free discipline: Stat arb requires holding losing positions through temporary spread widening — the strategy only works if you stay in the trade while it looks worst. Humans close losing trades too early under psychological pressure. Agents execute the rule exactly as written.
• Universe scanning: A human can monitor perhaps 5 pairs actively. An agent can scan 500 pairs simultaneously, running cointegration tests on each with fresh data every hour, and dynamically rotating into the best opportunities as old pairs decay.
• Record-keeping: An agent can log every trade, compute real-time Sharpe ratios, track spread half-lives, and report to a parent orchestrator — all without any human administrative overhead.

Cointegration Testing in Python

Before trading any pair, you must establish statistical evidence of cointegration. The Engle-Granger two-step procedure is the standard approach for two-asset pairs. First, regress one price series on the other using ordinary least squares. Second, apply the Augmented Dickey-Fuller test to the regression residuals. If the residuals are stationary (p-value < 0.05), the pair is cointegrated at the 95% confidence level.

import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import coint, adfuller
from statsmodels.regression.linear_model import OLS
from statsmodels.tools import add_constant
import requests
import logging

logger = logging.getLogger("stat_arb")

# Purple Flea Trading API base
API_BASE = "https://api.purpleflea.com/trading"
API_KEY  = "pf_live_your_key_here"

def fetch_closes(market: str, limit: int = 500) -> pd.Series:
    """Fetch hourly closing prices from Purple Flea Trading API."""
    r = requests.get(
        f"{API_BASE}/candles",
        params={"market": market, "interval": "1h", "limit": limit},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10,
    )
    r.raise_for_status()
    candles = r.json()["candles"]
    closes = pd.Series(
        [float(c["close"]) for c in candles],
        index=pd.to_datetime([c["time"] for c in candles], unit="ms"),
        name=market,
    )
    return closes

def test_cointegration(s1: pd.Series, s2: pd.Series) -> dict:
    """
    Run Engle-Granger cointegration test and return hedge ratio + p-value.
    Returns dict with keys: pvalue, hedge_ratio, cointegrated.
    """
    # Align on common index
    df = pd.concat([s1, s2], axis=1).dropna()
    y = df.iloc[:, 0]
    x = df.iloc[:, 1]

    # OLS regression to find hedge ratio
    ols = OLS(y, add_constant(x)).fit()
    hedge_ratio = ols.params[1]
    intercept   = ols.params[0]
    residuals   = y - (hedge_ratio * x + intercept)

    # ADF test on residuals
    adf_stat, pvalue, *_ = adfuller(residuals, autolag="AIC")

    return {
        "pvalue": round(pvalue, 4),
        "hedge_ratio": round(hedge_ratio, 6),
        "intercept": round(intercept, 6),
        "cointegrated": pvalue < 0.05,
        "adf_stat": round(adf_stat, 4),
    }

The Engle-Granger test is directional — regressing A on B gives a different hedge ratio than regressing B on A. Best practice is to run both directions and take the pair with the lower p-value, or to use the Johansen test for symmetric cointegration with multiple assets.

Purple Flea Integration

Purple Flea provides two services that together form a complete stat arb infrastructure for agents. The Trading API handles order placement, fills, and position management across 275+ perpetual markets. The Escrow service handles capital ring-fencing — you lock your risk budget into a trustless escrow contract at the start of a trading session, and automatically release profits to your wallet when positions close profitably.

This separation of concerns is important for agent risk management. An agent that conflates its trading capital with its general wallet balance may over-allocate to a single strategy. Escrow forces the agent to commit a defined risk budget upfront, and the 15% referral fee on the escrow contract can be structured to reward a supervising orchestrator agent that monitors the stat arb agent's risk compliance.

StatArbAgent: Complete Implementation

The following class implements a full stat arb agent. It scans a universe of markets to find cointegrated pairs, computes real-time spread z-scores, generates entry and exit signals, and routes execution through the Purple Flea Trading API. Every trade's risk capital is pulled from a pre-funded escrow contract.

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

@dataclass
class Pair:
    market_a:     str
    market_b:     str
    hedge_ratio:  float
    intercept:    float
    spread_mean:  float = 0.0
    spread_std:   float = 1.0
    position:     str   = "flat"  # "long_a", "short_a", "flat"
    entry_z:      float = 0.0
    pnl_usd:      float = 0.0

class StatArbAgent:
    """
    Full-cycle statistical arbitrage agent for Purple Flea perpetuals.
    Uses cointegration-based signals and Escrow for risk capital management.
    """

    UNIVERSE = [
        "BTC-USDC", "ETH-USDC", "SOL-USDC", "AVAX-USDC",
        "LINK-USDC", "MATIC-USDC", "ARB-USDC", "OP-USDC",
        "INJ-USDC", "TIA-USDC", "SEI-USDC", "SUI-USDC",
    ]

    def __init__(
        self,
        api_key:       str,
        escrow_id:     str,
        wallet_id:     str,
        entry_z:       float = 2.0,
        exit_z:        float = 0.5,
        stop_z:        float = 3.5,
        max_pairs:     int   = 5,
        notional_usd:  float = 50.0,
    ):
        self.api_key      = api_key
        self.escrow_id    = escrow_id
        self.wallet_id    = wallet_id
        self.entry_z      = entry_z
        self.exit_z       = exit_z
        self.stop_z       = stop_z
        self.max_pairs    = max_pairs
        self.notional_usd = notional_usd
        self.pairs: list[Pair] = []
        self.client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=15,
        )

    # ─── Pair Discovery ────────────────────────────────────────────────────────

    async def find_pairs(self) -> list[Pair]:
        """
        Scan universe for cointegrated pairs.
        Returns up to max_pairs sorted by cointegration p-value.
        """
        closes: dict[str, pd.Series] = {}
        for market in self.UNIVERSE:
            try:
                closes[market] = fetch_closes(market, limit=300)
            except Exception as e:
                logger.warning(f"Failed to fetch {market}: {e}")

        markets = list(closes.keys())
        candidates = []

        for i in range(len(markets)):
            for j in range(i + 1, len(markets)):
                ma, mb = markets[i], markets[j]
                result = test_cointegration(closes[ma], closes[mb])
                if result["cointegrated"]:
                    # Compute historical spread stats
                    s1 = closes[ma]
                    s2 = closes[mb]
                    df = pd.concat([s1, s2], axis=1).dropna()
                    spread = df.iloc[:, 0] - result["hedge_ratio"] * df.iloc[:, 1] - result["intercept"]
                    candidates.append({
                        "pair": Pair(
                            market_a=ma,
                            market_b=mb,
                            hedge_ratio=result["hedge_ratio"],
                            intercept=result["intercept"],
                            spread_mean=float(spread.mean()),
                            spread_std=float(spread.std()),
                        ),
                        "pvalue": result["pvalue"],
                    })

        candidates.sort(key=lambda x: x["pvalue"])
        self.pairs = [c["pair"] for c in candidates[:self.max_pairs]]
        logger.info(f"Found {len(self.pairs)} cointegrated pairs")
        return self.pairs

    # ─── Spread & Signal ───────────────────────────────────────────────────────

    async def compute_spread(self, pair: Pair) -> float:
        """Fetch latest mark prices and compute current spread z-score."""
        r = await self.client.get(
            f"https://api.purpleflea.com/trading/tickers",
            params={"markets": f"{pair.market_a},{pair.market_b}"},
        )
        r.raise_for_status()
        tickers = {t["market"]: float(t["mark_price"]) for t in r.json()["tickers"]}
        pa = tickers[pair.market_a]
        pb = tickers[pair.market_b]
        spread = pa - pair.hedge_ratio * pb - pair.intercept
        z_score = (spread - pair.spread_mean) / (pair.spread_std + 1e-9)
        return z_score

    def generate_signal(self, pair: Pair, z: float) -> str:
        """
        Generate trading signal from current z-score and pair state.
        Returns: "buy_spread", "sell_spread", "exit", "stop", "hold"
        """
        if pair.position == "flat":
            if z > self.entry_z:
                return "sell_spread"   # spread too high → expect reversion down
            if z < -self.entry_z:
                return "buy_spread"    # spread too low → expect reversion up
            return "hold"
        # Position already open — check exit or stop conditions
        if abs(z) < self.exit_z:
            return "exit"
        if abs(z) > self.stop_z:
            return "stop"         # stop-loss: correlation breakdown
        return "hold"

    # ─── Execution ─────────────────────────────────────────────────────────────

    async def execute_trade(self, pair: Pair, signal: str) -> dict:
        """
        Execute both legs of a spread trade via Purple Flea Trading API.
        Draws margin from escrow contract.
        """
        if signal == "buy_spread":
            # Long market_a, short market_b (scaled by hedge_ratio)
            leg_a = {"market": pair.market_a, "side": "buy",  "notional": self.notional_usd}
            leg_b = {"market": pair.market_b, "side": "sell", "notional": self.notional_usd * pair.hedge_ratio}
        elif signal in ("sell_spread",):
            leg_a = {"market": pair.market_a, "side": "sell", "notional": self.notional_usd}
            leg_b = {"market": pair.market_b, "side": "buy",  "notional": self.notional_usd * pair.hedge_ratio}
        else:
            # exit / stop — close both positions
            leg_a = {"market": pair.market_a, "side": "close", "notional": 0}
            leg_b = {"market": pair.market_b, "side": "close", "notional": 0}

        payload = {
            "legs": [leg_a, leg_b],
            "escrow_id": self.escrow_id,
            "order_type": "market",
            "reduce_only": signal in ("exit", "stop"),
        }

        r = await self.client.post(
            "https://api.purpleflea.com/trading/orders/multi",
            json=payload,
        )
        r.raise_for_status()
        result = r.json()
        logger.info(f"Executed {signal} on {pair.market_a}/{pair.market_b}: {result}")

        # Update pair state
        if signal == "buy_spread":
            pair.position = "long_a"
        elif signal == "sell_spread":
            pair.position = "short_a"
        else:
            pair.position = "flat"

        return result

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

    async def run(self, interval_seconds: int = 300):
        """
        Main trading loop. Re-scans for pairs every 24 hours,
        evaluates signals every interval_seconds.
        """
        logger.info("StatArbAgent starting — scanning for pairs...")
        await self.find_pairs()
        last_scan = time.time()

        while True:
            # Re-scan universe every 24 hours
            if time.time() - last_scan > 86400:
                await self.find_pairs()
                last_scan = time.time()

            for pair in self.pairs:
                try:
                    z = await self.compute_spread(pair)
                    signal = self.generate_signal(pair, z)
                    if signal != "hold":
                        await self.execute_trade(pair, signal)
                except Exception as e:
                    logger.error(f"Error on pair {pair.market_a}/{pair.market_b}: {e}")

            await asyncio.sleep(interval_seconds)

Capital Management with Escrow

Capital management is where most retail stat arb implementations fail. Without explicit allocation controls, an agent can pyramid into losing positions, allocate the same capital to multiple strategies simultaneously, or fail to take profits when a trade closes successfully. Purple Flea Escrow solves all three problems.

The recommended pattern is to create one escrow contract per pair at session start. The contract holds the maximum risk capital for that pair — for example, 100 USDC for a pair that might lose 20% at a 3.5 z-score stop. When a trade closes profitably, the profit is released to the agent's main wallet automatically via the escrow settlement. When a stop-loss fires, only the escrowed capital is at risk — the agent's broader wallet is protected.

async def setup_escrow_contracts(pairs: list[Pair], risk_per_pair_usdc: float, api_key: str) -> dict[str, str]:
    """
    Create one escrow contract per trading pair.
    Returns dict mapping pair identifier to escrow_id.
    """
    escrow_map = {}
    async with httpx.AsyncClient(headers={"Authorization": f"Bearer {api_key}"}) as client:
        for pair in pairs:
            pair_id = f"{pair.market_a}_{pair.market_b}"
            r = await client.post(
                "https://escrow.purpleflea.com/api/create",
                json={
                    "amount_usdc": risk_per_pair_usdc,
                    "purpose": f"stat_arb_{pair_id}",
                    "auto_release": True,   # release profit to wallet on settlement
                    "referral_bps": 1500,      # 15% referral to orchestrator
                },
            )
            r.raise_for_status()
            escrow_id = r.json()["escrow_id"]
            escrow_map[pair_id] = escrow_id
            logger.info(f"Created escrow {escrow_id} for {pair_id} ({risk_per_pair_usdc} USDC)")
    return escrow_map

Backtesting Framework

Before deploying any stat arb strategy with real capital, rigorous backtesting is mandatory. The spread half-life — how long it takes the spread to decay by half toward its mean — determines how frequently you can expect profitable trades and what holding period to budget for. A spread with a 12-hour half-life is very different from one with a 7-day half-life.

import numpy as np
import pandas as pd

def compute_half_life(spread: pd.Series) -> float:
    """
    Estimate half-life of mean reversion via OLS of spread on its lag.
    half_life = -log(2) / log(1 + phi) where phi is the AR(1) coefficient.
    """
    spread_lag = spread.shift(1)
    delta = spread - spread_lag
    df = pd.DataFrame({"delta": delta, "lag": spread_lag}).dropna()
    phi = np.linalg.lstsq(df[["lag"]], df["delta"], rcond=None)[0][0]
    half_life = -np.log(2) / np.log(1 + phi) if phi < 0 else np.inf
    return half_life

def backtest_pair(
    prices_a: pd.Series,
    prices_b: pd.Series,
    hedge_ratio: float,
    intercept: float,
    entry_z: float = 2.0,
    exit_z:  float = 0.5,
    stop_z:  float = 3.5,
    notional: float = 1000.0,
    fee_bps: float = 5,      # 0.05% per leg
) -> dict:
    """
    Event-driven backtest of a single spread strategy.
    Returns performance metrics dict.
    """
    df = pd.concat([prices_a, prices_b], axis=1).dropna()
    df.columns = ["A", "B"]
    spread = df["A"] - hedge_ratio * df["B"] - intercept
    z = (spread - spread.mean()) / (spread.std() + 1e-9)

    position = 0   # +1 = long spread, -1 = short spread
    trades, equity_curve = [], [0.0]
    entry_z_score = 0.0

    for i in range(1, len(z)):
        zi = z.iloc[i]
        fee_cost = 2 * notional * fee_bps / 10000   # two legs

        if position == 0:
            if zi > entry_z:
                position, entry_z_score = -1, zi
            elif zi < -entry_z:
                position, entry_z_score = 1, zi
        elif (abs(zi) < exit_z) or (abs(zi) > stop_z):
            pnl = position * (spread.iloc[i] - spread.iloc[i - 1]) * notional / spread.iloc[i - 1]
            trades.append(pnl - fee_cost)
            equity_curve.append(equity_curve[-1] + pnl - fee_cost)
            position = 0
        else:
            # Mark-to-market P&L
            pnl = position * (spread.iloc[i] - spread.iloc[i - 1]) * notional / spread.iloc[i - 1]
            equity_curve.append(equity_curve[-1] + pnl)

    equity = pd.Series(equity_curve)
    returns = equity.diff().dropna()
    sharpe = returns.mean() / (returns.std() + 1e-9) * np.sqrt(8760)  # annualized hourly
    max_dd = (equity - equity.cummax()).min()

    return {
        "total_pnl_usd": round(equity.iloc[-1], 2),
        "n_trades": len(trades),
        "sharpe": round(sharpe, 2),
        "max_drawdown_usd": round(max_dd, 2),
        "win_rate": round(sum(t > 0 for t in trades) / (len(trades) + 1), 3),
        "half_life_hours": round(compute_half_life(spread), 1),
    }

Risk Controls

Every stat arb system needs multiple layers of risk control. The primary risk is not that the strategy loses money on individual trades — it is that the cointegration relationship breaks down permanently and the spread never reverts. This is called a correlation breakdown or regime change, and it can happen when the fundamentals of one asset change (exchange hack, protocol fork, regulatory action).

async def check_correlation_health(pair: Pair, window: int = 60) -> bool:
    """
    Rolling cointegration check — returns False if pair should be retired.
    """
    s1 = fetch_closes(pair.market_a, limit=window)
    s2 = fetch_closes(pair.market_b, limit=window)
    result = test_cointegration(s1, s2)

    # Check rolling 60-period correlation
    df = pd.concat([s1, s2], axis=1).dropna()
    rolling_corr = df.iloc[:, 0].corr(df.iloc[:, 1])

    healthy = result["pvalue"] < 0.10 and rolling_corr > 0.60
    if not healthy:
        logger.warning(
            f"Pair {pair.market_a}/{pair.market_b} failing health check: "
            f"p={result['pvalue']:.3f}, corr={rolling_corr:.3f}"
        )
    return healthy

Live P&L Tracking via Purple Flea Wallet

Purple Flea Wallet provides a real-time ledger for every agent account. After each trade closes, the stat arb agent should record the closed P&L against the relevant escrow contract. The wallet API supports custom metadata tagging — tag each entry with the pair identifier, z-score at entry, and hold time so you can aggregate statistics by pair over time.

async def record_pnl(
    wallet_id: str,
    pair: Pair,
    pnl_usd: float,
    hold_hours: float,
    entry_z: float,
    exit_z: float,
    api_key: str,
):
    """Record closed trade P&L to Purple Flea Wallet with metadata."""
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"https://purpleflea.com/api/wallet/{wallet_id}/ledger",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "amount_usd": pnl_usd,
                "type": "trade_pnl",
                "metadata": {
                    "strategy": "stat_arb",
                    "pair": f"{pair.market_a}/{pair.market_b}",
                    "entry_z": entry_z,
                    "exit_z": exit_z,
                    "hold_hours": hold_hours,
                    "hedge_ratio": pair.hedge_ratio,
                },
            },
        )
        r.raise_for_status()

With consistent tagging, you can run weekly reports across all stat arb positions: which pairs are performing, which have decaying cointegration, what the average hold time is, and whether the realized Sharpe is tracking the backtest estimate. This feedback loop is what separates a disciplined stat arb program from speculative spread betting.

Performance Metrics

Evaluating a stat arb agent requires metrics that reflect the market-neutral nature of the strategy. Simple absolute returns are insufficient — you need risk-adjusted measures that account for the fact that a good stat arb strategy should have low correlation to market beta.

def compute_performance_metrics(trades: list[float], equity_curve: list[float], btc_returns: pd.Series) -> dict:
    """Compute full performance metric suite for a stat arb book."""
    equity = pd.Series(equity_curve)
    returns = equity.diff().dropna()

    # Sharpe (annualized hourly)
    sharpe = returns.mean() / (returns.std() + 1e-9) * np.sqrt(8760)

    # Max drawdown
    running_max = equity.cummax()
    drawdowns = (equity - running_max) / (running_max + 1e-9)
    max_dd = drawdowns.min()

    # Calmar
    ann_return = returns.sum() * (8760 / len(returns))
    calmar = ann_return / (abs(max_dd) + 1e-9)

    # Market beta — align returns with btc_returns by length
    min_len = min(len(returns), len(btc_returns))
    r_aligned = returns.iloc[-min_len:]
    btc_aligned = btc_returns.iloc[-min_len:]
    cov = np.cov(r_aligned, btc_aligned)
    beta = cov[0, 1] / (cov[1, 1] + 1e-9)

    return {
        "sharpe": round(sharpe, 2),
        "max_drawdown_pct": round(max_dd * 100, 2),
        "calmar": round(calmar, 2),
        "win_rate": round(sum(t > 0 for t in trades) / (len(trades) + 1), 3),
        "market_beta": round(beta, 4),
        "ann_return_pct": round(ann_return * 100, 2),
        "n_trades": len(trades),
    }

Next Steps

Statistical arbitrage is a deep field with many extensions beyond simple two-asset pair trading. Once the basic pipeline is working, consider these enhancements:

• Johansen test: For trading baskets of three or more assets with multiple cointegrating relationships — useful for sector-neutral strategies across DeFi tokens.
• Dynamic hedge ratio: Instead of a static OLS hedge ratio, use a Kalman filter to track the evolving relationship between two assets in real time. This is more robust to regime shifts.
• Half-life filtering: Only trade pairs where the estimated half-life is between 2 and 48 hours. Too-short half-lives produce excessive fees; too-long half-lives tie up capital for too long.
• Multi-strategy netting: Use a single escrow contract across all pairs with automatic rebalancing — pairs that close profitably fund new opportunities without requiring external capital top-ups.

Purple Flea provides the complete execution infrastructure for all of these patterns. The Trading API, Escrow, and Wallet together form a production-grade foundation that any agent can build a stat arb program on top of today.