Basis Convergence Trading
for AI Agents

Basis convergence is one of the most reliable and scalable arbitrage strategies available to autonomous agents in crypto markets. When the spot price and perpetual futures price diverge, market forces guarantee they will eventually reconverge — and a well-designed agent can systematically capture that spread. This guide covers every layer: the mechanics of basis formation, calendar spreads, cash-and-carry execution, risk controls, and a complete BasisConvergenceAgent class built on the Purple Flea trading API.

1. What Is Basis — and Why Does It Diverge?

In derivatives markets, basis is the price difference between a derivative and its underlying spot asset. For cryptocurrency perpetual futures — the dominant instrument on Purple Flea and every major crypto exchange — basis manifests as:

In traditional finance, futures have fixed expiries. Basis collapses to zero at settlement — that convergence is mechanical and guaranteed. Perpetual futures have no expiry date, so the convergence mechanism is different: the funding rate.

The Funding Rate as a Convergence Force

Every 8 hours (on most exchanges), long positions pay shorts when the perp trades above spot (positive basis / contango), and short positions pay longs when the perp trades below spot (negative basis / backwardation). This creates a continuous economic incentive for the market to close the gap.

Primary Drivers of Basis Formation

• Leverage demand: During bull markets, traders crave leveraged long exposure through perps, bidding the perp premium up.
• Spot supply constraints: Limited spot market liquidity (especially in altcoins) creates pricing dislocations.
• Macro events: Fed announcements, regulatory news, and protocol upgrades cause directional bets through perps before spot markets adjust.
• Cross-exchange fragmentation: Different exchanges have different funding rate calculation windows, creating temporary multi-venue basis discrepancies.
• Collateral costs: For basis traders, the cost of tying up USD collateral in spot positions creates a minimum basis threshold below which convergence plays are unprofitable.

2. Types of Basis Convergence Strategies

Basis convergence encompasses a family of related strategies. Agents should understand all of them and deploy whichever is optimal given current market conditions, volatility regime, and available capital.

2.1 Cash-and-Carry (Spot-Long / Perp-Short)

The canonical basis trade. When the perpetual trades at a premium to spot, an agent buys spot (going long the underlying) and simultaneously shorts the equivalent notional in the perpetual. The resulting portfolio is delta-neutral: spot price moves cancel out. The agent captures the basis compression over time plus the funding payments received for holding the short position.

This is a convergence play in two senses: (1) the basis itself may compress as the premium erodes, and (2) the agent receives ongoing funding payments as long as the perp stays in contango. The trade is theoretically risk-free if executed perfectly — but in practice, liquidation risk, slippage, and borrow costs require careful risk management.

2.2 Reverse Cash-and-Carry (Spot-Short / Perp-Long)

When the perpetual trades at a discount to spot (backwardation), the trade inverts: borrow and short spot, go long the perp. The agent captures the negative basis as it reverts toward zero and collects funding from short-payers. This is less common in crypto (backwardation is rarer and shorter-lived) but can be very profitable during extreme fear events.

2.3 Calendar Spread (Perp vs. Quarterly Futures)

On exchanges that offer both perpetual and quarterly futures (e.g., CME, Deribit, some CEXes), agents can trade the spread between the two. Quarterly futures have hard settlement — their basis must converge to zero at expiry. This creates a predictable convergence timeline that perpetual basis lacks.

This is a lower-volatility basis play: the agent is insulated from directional price moves because both legs move together. The P&L driver is purely the spread compression as expiry approaches (the "roll-down" of the quarterly's basis).

2.4 Cross-Exchange Basis Arbitrage

Different exchanges price the same perpetual differently due to their independent funding rate formulas, liquidity pools, and user bases. An agent can go long on the exchange with the lower perp price and short on the exchange with the higher perp price. Both legs converge toward the true spot price independently.

3. Measuring and Monitoring Basis

Before an agent can trade basis convergence, it must accurately and continuously measure the current basis across all relevant pairs. Raw basis measurement is straightforward; the challenge lies in building meaningful signals from that data.

Raw Basis Calculation

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

PURPLE_FLEA_BASE = "https://api.purpleflea.com/v1"
HEADERS = {"Authorization": "Bearer pf_live_YOUR_KEY_HERE"}

@dataclass
class BasisSnapshot:
    symbol: str
    spot_price: float
    perp_price: float
    raw_basis: float          # perp - spot (absolute)
    basis_pct: float          # basis / spot * 100
    funding_rate_8h: float    # current 8h funding rate
    annualized_basis_rate: float  # basis_pct * (365 * 3)  (3 funding intervals/day)
    timestamp: datetime = field(default_factory=datetime.utcnow)

    @property
    def contango(self) -> bool:
        return self.raw_basis > 0

    @property
    def backwardation(self) -> bool:
        return self.raw_basis < 0

    def __repr__(self):
        direction = "CONTANGO" if self.contango else "BACKWARDATION"
        return (
            f"{self.symbol} [{direction}] "
            f"Basis={self.basis_pct:+.4f}% "
            f"Funding={self.funding_rate_8h*100:+.4f}%/8h "
            f"AnnRate={self.annualized_basis_rate:.1f}%"
        )


async def fetch_basis_snapshot(client: httpx.AsyncClient, symbol: str) -> Optional[BasisSnapshot]:
    """Fetch spot and perp prices, calculate basis."""
    try:
        spot_resp, perp_resp, funding_resp = await asyncio.gather(
            client.get(f"{PURPLE_FLEA_BASE}/spot/ticker/{symbol}USDT", headers=HEADERS),
            client.get(f"{PURPLE_FLEA_BASE}/perp/ticker/{symbol}USDT-PERP", headers=HEADERS),
            client.get(f"{PURPLE_FLEA_BASE}/perp/funding/{symbol}USDT-PERP", headers=HEADERS),
        )
        spot_price = float(spot_resp.json()["last_price"])
        perp_price = float(perp_resp.json()["mark_price"])
        funding_rate = float(funding_resp.json()["current_rate"])

        raw_basis = perp_price - spot_price
        basis_pct = (raw_basis / spot_price) * 100
        annualized = basis_pct * (365 * 3)   # 3 x 8h intervals per day

        return BasisSnapshot(
            symbol=symbol,
            spot_price=spot_price,
            perp_price=perp_price,
            raw_basis=raw_basis,
            basis_pct=basis_pct,
            funding_rate_8h=funding_rate,
            annualized_basis_rate=annualized,
        )
    except Exception as e:
        print(f"[BasisMonitor] Error fetching {symbol}: {e}")
        return None


async def scan_basis_universe(symbols: list[str]) -> list[BasisSnapshot]:
    """Scan multiple symbols for basis opportunities."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        snapshots = await asyncio.gather(*[fetch_basis_snapshot(client, s) for s in symbols])
    return [s for s in snapshots if s is not None]


# Example usage
async def main():
    symbols = ["BTC", "ETH", "SOL", "ARB", "AVAX", "DOGE"]
    snapshots = await scan_basis_universe(symbols)

    print("\n=== BASIS UNIVERSE SCAN ===")
    for snap in sorted(snapshots, key=lambda x: abs(x.basis_pct), reverse=True):
        print(snap)

asyncio.run(main())

Basis Z-Score: Normalizing for Comparison

Raw basis percentages vary by asset (BTC basis behaves differently from altcoin basis). To compare opportunities across assets on a fair footing, normalize using a rolling z-score:

import numpy as np
from collections import deque

class BasisZScoreTracker:
    """Track rolling basis stats and compute z-scores for opportunity detection."""

    def __init__(self, window_size: int = 1080):  # 1080 = 30 days of 8h samples
        self.window_size = window_size
        self._histories: dict[str, deque] = {}

    def update(self, symbol: str, basis_pct: float) -> float | None:
        """Add new basis observation. Returns z-score if enough history."""
        if symbol not in self._histories:
            self._histories[symbol] = deque(maxlen=self.window_size)

        hist = self._histories[symbol]
        hist.append(basis_pct)

        if len(hist) < 30:  # Need at least 30 observations for meaningful stats
            return None

        arr = np.array(hist)
        mean = arr.mean()
        std = arr.std()
        if std < 1e-8:
            return 0.0

        return (basis_pct - mean) / std

    def get_stats(self, symbol: str) -> dict:
        if symbol not in self._histories or len(self._histories[symbol]) < 10:
            return {}
        arr = np.array(self._histories[symbol])
        return {
            "mean": arr.mean(),
            "std": arr.std(),
            "min": arr.min(),
            "max": arr.max(),
            "percentile_90": np.percentile(arr, 90),
            "percentile_10": np.percentile(arr, 10),
        }

4. Full BasisConvergenceAgent Implementation

The following class encapsulates a complete basis convergence trading agent. It scans multiple assets, detects opportunities using z-scores, executes delta-neutral entries, monitors funding payments, and exits when the basis reverts.

import asyncio
import httpx
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
from basis_monitor import scan_basis_universe, BasisSnapshot
from basis_zscore import BasisZScoreTracker

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("BasisAgent")

API_BASE = "https://api.purpleflea.com/v1"
API_KEY = "pf_live_YOUR_KEY_HERE"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

# ─── Configuration ────────────────────────────────────────────────────────────

@dataclass
class BasisAgentConfig:
    symbols: list[str] = field(default_factory=lambda: ["BTC", "ETH", "SOL", "ARB"])
    min_basis_pct: float = 0.25          # Minimum basis (%) to enter (annualized ~9%)
    min_zscore: float = 1.8              # Minimum z-score to consider entry
    max_position_usd: float = 5_000.0   # Max notional per trade
    total_capital_usd: float = 50_000.0 # Total agent capital
    max_leverage_perp: float = 3.0      # Max leverage on perp short leg
    scan_interval_sec: int = 60          # How often to scan universe
    funding_interval_sec: int = 28800    # 8 hours in seconds
    exit_basis_pct: float = 0.05         # Exit when basis narrows to this %
    exit_zscore: float = 0.3             # Or when z-score reverts to this
    max_hold_days: int = 30              # Force exit after this many days
    slippage_buffer_pct: float = 0.05    # Assume 0.05% slippage per leg
    fee_per_side_pct: float = 0.04       # Taker fee (both legs)

# ─── Position Tracking ────────────────────────────────────────────────────────

@dataclass
class BasisPosition:
    symbol: str
    spot_qty: float           # Units of spot held long
    perp_qty: float           # Units of perp short
    spot_entry_price: float
    perp_entry_price: float
    entry_basis_pct: float
    notional_usd: float
    entered_at: datetime = field(default_factory=datetime.utcnow)
    spot_order_id: str = ""
    perp_order_id: str = ""
    funding_received_usd: float = 0.0

    @property
    def age_hours(self) -> float:
        return (datetime.utcnow() - self.entered_at).total_seconds() / 3600

    @property
    def target_exit_basis_pct(self) -> float:
        return self.entry_basis_pct * 0.15  # Exit when 85% of basis captured

# ─── Agent Core ───────────────────────────────────────────────────────────────

class BasisConvergenceAgent:
    def __init__(self, config: BasisAgentConfig):
        self.config = config
        self.z_tracker = BasisZScoreTracker(window_size=1080)
        self.positions: dict[str, BasisPosition] = {}
        self.client: Optional[httpx.AsyncClient] = None
        self._running = False
        self._total_pnl_usd = 0.0
        self._trade_count = 0

    # ── API Helpers ───────────────────────────────────────────────────────────

    async def _place_spot_buy(self, symbol: str, usd_amount: float) -> dict:
        ticker = f"{symbol}USDT"
        payload = {
            "symbol": ticker,
            "side": "buy",
            "type": "market",
            "quote_amount": usd_amount,
        }
        resp = await self.client.post(f"{API_BASE}/spot/order", json=payload, headers=HEADERS)
        resp.raise_for_status()
        return resp.json()

    async def _place_perp_short(self, symbol: str, usd_amount: float, leverage: float) -> dict:
        ticker = f"{symbol}USDT-PERP"
        payload = {
            "symbol": ticker,
            "side": "sell",
            "type": "market",
            "quote_amount": usd_amount,
            "leverage": leverage,
            "reduce_only": False,
        }
        resp = await self.client.post(f"{API_BASE}/perp/order", json=payload, headers=HEADERS)
        resp.raise_for_status()
        return resp.json()

    async def _close_spot_position(self, symbol: str, qty: float) -> dict:
        ticker = f"{symbol}USDT"
        payload = {"symbol": ticker, "side": "sell", "type": "market", "base_amount": qty}
        resp = await self.client.post(f"{API_BASE}/spot/order", json=payload, headers=HEADERS)
        resp.raise_for_status()
        return resp.json()

    async def _close_perp_short(self, symbol: str, qty: float) -> dict:
        ticker = f"{symbol}USDT-PERP"
        payload = {
            "symbol": ticker,
            "side": "buy",
            "type": "market",
            "base_amount": qty,
            "reduce_only": True,
        }
        resp = await self.client.post(f"{API_BASE}/perp/order", json=payload, headers=HEADERS)
        resp.raise_for_status()
        return resp.json()

    async def _get_funding_received(self, symbol: str, since: datetime) -> float:
        ticker = f"{symbol}USDT-PERP"
        resp = await self.client.get(
            f"{API_BASE}/perp/funding-history",
            params={"symbol": ticker, "since": since.isoformat()},
            headers=HEADERS,
        )
        if resp.status_code != 200:
            return 0.0
        events = resp.json().get("events", [])
        # Negative payment = we received (as short when funding is positive)
        return sum(-e["payment_usd"] for e in events)

    # ── Strategy Logic ────────────────────────────────────────────────────────

    def _net_entry_cost_pct(self) -> float:
        """Total round-trip cost as percentage."""
        fees = self.config.fee_per_side_pct * 2 * 2      # 2 legs, entry + exit
        slippage = self.config.slippage_buffer_pct * 2 * 2
        return fees + slippage

    def _calculate_net_apy(self, basis_pct: float, funding_rate_8h: float) -> float:
        """
        Estimate net annualized return from basis + funding, minus costs.
        basis_pct: current basis as % of spot
        funding_rate_8h: rate per 8-hour period (as decimal, e.g. 0.0001)
        """
        basis_annual = basis_pct * 3 * 365         # Annualized basis capture
        funding_annual = funding_rate_8h * 3 * 365 * 100  # Annualized funding %
        cost_annual = self._net_entry_cost_pct() * 12  # Assume 12 rotations/year
        return basis_annual + funding_annual - cost_annual

    def _capital_available(self) -> float:
        used = sum(p.notional_usd for p in self.positions.values())
        return max(0.0, self.config.total_capital_usd - used)

    def _position_size(self, opportunity_score: float) -> float:
        """Size position by opportunity score (z-score), capped at max."""
        base = min(self._capital_available() * 0.20, self.config.max_position_usd)
        scale = min(opportunity_score / 3.0, 1.5)  # Scale up to 1.5x for z > 3
        return min(base * scale, self.config.max_position_usd)

    # ── Scan and Detect ───────────────────────────────────────────────────────

    async def scan_and_rank_opportunities(self) -> list[tuple[BasisSnapshot, float]]:
        """Return (snapshot, z_score) pairs sorted by opportunity score."""
        snapshots = await scan_basis_universe(self.config.symbols)
        opportunities = []

        for snap in snapshots:
            z = self.z_tracker.update(snap.symbol, snap.basis_pct)
            if z is None:
                continue

            # Only care about meaningful positive contango (cash-and-carry)
            if snap.basis_pct < self.config.min_basis_pct:
                continue
            if z < self.config.min_zscore:
                continue
            if snap.symbol in self.positions:
                continue  # Already have a position

            net_apy = self._calculate_net_apy(snap.basis_pct, snap.funding_rate_8h)
            if net_apy < 5.0:  # Require at least 5% net APY
                log.info(f"[{snap.symbol}] Skipping: net APY {net_apy:.1f}% too low")
                continue

            log.info(f"[{snap.symbol}] Opportunity: z={z:.2f}, basis={snap.basis_pct:.4f}%, netAPY={net_apy:.1f}%")
            opportunities.append((snap, z))

        return sorted(opportunities, key=lambda x: x[1], reverse=True)

    # ── Entry ─────────────────────────────────────────────────────────────────

    async def enter_basis_trade(self, snap: BasisSnapshot, z_score: float) -> bool:
        """Execute a delta-neutral cash-and-carry basis trade."""
        symbol = snap.symbol
        notional = self._position_size(z_score)

        if notional < 100:
            log.warning(f"[{symbol}] Insufficient capital for entry (${notional:.0f})")
            return False

        log.info(f"[{symbol}] Entering basis trade: notional=${notional:.0f}, basis={snap.basis_pct:.4f}%")

        try:
            # Place both legs concurrently
            spot_result, perp_result = await asyncio.gather(
                self._place_spot_buy(symbol, notional),
                self._place_perp_short(symbol, notional, self.config.max_leverage_perp),
            )

            spot_fill = float(spot_result["avg_fill_price"])
            perp_fill = float(perp_result["avg_fill_price"])
            spot_qty = float(spot_result["filled_base"])
            perp_qty = float(perp_result["filled_base"])

            position = BasisPosition(
                symbol=symbol,
                spot_qty=spot_qty,
                perp_qty=perp_qty,
                spot_entry_price=spot_fill,
                perp_entry_price=perp_fill,
                entry_basis_pct=snap.basis_pct,
                notional_usd=notional,
                spot_order_id=spot_result["order_id"],
                perp_order_id=perp_result["order_id"],
            )

            self.positions[symbol] = position
            self._trade_count += 1
            log.info(f"[{symbol}] Position opened. SpotFill={spot_fill:.2f}, PerpFill={perp_fill:.2f}")
            return True

        except Exception as e:
            log.error(f"[{symbol}] Entry failed: {e}")
            return False

    # ── Exit ──────────────────────────────────────────────────────────────────

    async def exit_basis_trade(self, symbol: str, snap: BasisSnapshot, reason: str) -> None:
        """Close both legs of the basis trade."""
        pos = self.positions.get(symbol)
        if not pos:
            return

        log.info(f"[{symbol}] Exiting basis trade. Reason: {reason}")

        try:
            # Fetch accumulated funding before closing
            funding_received = await self._get_funding_received(symbol, pos.entered_at)
            pos.funding_received_usd = funding_received

            # Close both legs concurrently
            await asyncio.gather(
                self._close_spot_position(symbol, pos.spot_qty),
                self._close_perp_short(symbol, pos.perp_qty),
            )

            # Calculate P&L
            basis_captured_pct = pos.entry_basis_pct - snap.basis_pct
            basis_pnl = pos.notional_usd * (basis_captured_pct / 100)
            total_pnl = basis_pnl + funding_received
            self._total_pnl_usd += total_pnl

            log.info(
                f"[{symbol}] CLOSED. BasisPnL=${basis_pnl:.2f}, "
                f"Funding=${funding_received:.2f}, Total=${total_pnl:.2f}, "
                f"HeldFor={pos.age_hours:.1f}h"
            )

            del self.positions[symbol]

        except Exception as e:
            log.error(f"[{symbol}] Exit failed: {e}")

    # ── Monitor Existing Positions ────────────────────────────────────────────

    async def monitor_positions(self, snapshots: list[BasisSnapshot]) -> None:
        snap_by_symbol = {s.symbol: s for s in snapshots}

        for symbol, pos in list(self.positions.items()):
            snap = snap_by_symbol.get(symbol)
            if not snap:
                continue

            z = self.z_tracker.update(symbol, snap.basis_pct)

            # Exit conditions
            if snap.basis_pct <= self.config.exit_basis_pct:
                await self.exit_basis_trade(symbol, snap, "basis_converged")
            elif z is not None and z <= self.config.exit_zscore:
                await self.exit_basis_trade(symbol, snap, "zscore_reverted")
            elif pos.age_hours > self.config.max_hold_days * 24:
                await self.exit_basis_trade(symbol, snap, "max_hold_exceeded")
            elif snap.basis_pct < 0:
                # Basis flipped — we're now being charged funding on the short
                await self.exit_basis_trade(symbol, snap, "basis_inverted")
            else:
                log.info(
                    f"[{symbol}] Position alive. Basis={snap.basis_pct:.4f}%, "
                    f"Z={z:.2f}, Age={pos.age_hours:.1f}h, "
                    f"FundingEst=${pos.notional_usd * snap.funding_rate_8h:.2f}/8h"
                )

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

    async def run(self) -> None:
        self._running = True
        log.info("BasisConvergenceAgent started.")

        async with httpx.AsyncClient(timeout=15.0) as client:
            self.client = client

            while self._running:
                try:
                    # Scan universe
                    opportunities = await self.scan_and_rank_opportunities()

                    # Monitor existing positions (needs current snapshots)
                    all_snaps = await scan_basis_universe(self.config.symbols)
                    await self.monitor_positions(all_snaps)

                    # Enter new opportunities (best first, within capital limits)
                    for snap, z in opportunities:
                        if self._capital_available() < 200:
                            break
                        await self.enter_basis_trade(snap, z)

                    log.info(
                        f"State: {len(self.positions)} positions, "
                        f"${self._capital_available():.0f} free, "
                        f"TotalPnL=${self._total_pnl_usd:.2f}"
                    )

                except asyncio.CancelledError:
                    break
                except Exception as e:
                    log.error(f"Main loop error: {e}")

                await asyncio.sleep(self.config.scan_interval_sec)

        self._running = False
        log.info(f"Agent stopped. Total trades: {self._trade_count}, P&L: ${self._total_pnl_usd:.2f}")

    def stop(self) -> None:
        self._running = False


# ─── Entry Point ──────────────────────────────────────────────────────────────

if __name__ == "__main__":
    config = BasisAgentConfig(
        symbols=["BTC", "ETH", "SOL", "ARB", "AVAX"],
        min_basis_pct=0.20,
        min_zscore=1.8,
        max_position_usd=10_000.0,
        total_capital_usd=100_000.0,
        scan_interval_sec=120,
    )
    agent = BasisConvergenceAgent(config)
    asyncio.run(agent.run())

5. Risk Management for Basis Trades

Basis convergence is often called a "market-neutral" strategy, but it is not risk-free. Understanding and actively managing each risk category is essential for an agent operating at scale.

5.1 Delta Neutrality and Maintenance

The core assumption of cash-and-carry is delta neutrality: gains on the spot leg offset losses on the short perp leg, and vice versa. But delta neutrality degrades over time:

• Liquidation on the perp short: If the price surges, the short position may approach liquidation before the spot leg gains are accessible. Always leave significant liquidation buffer (typically 50%+ of the margin requirement above the liquidation price).
• Spot leg loss of sync: If the perp short is partially liquidated, the spot leg becomes a directional long position. The agent must immediately sell spot proportionally.
• Perp margin drift: Unrealized gains/losses on the perp change the effective margin. Monitor margin ratio on every cycle.

async def check_delta_neutrality(client, position: BasisPosition, current_price: float) -> dict:
    """Verify delta neutrality. Returns adjustment recommendation."""
    spot_value_usd = position.spot_qty * current_price
    perp_short_value_usd = position.perp_qty * current_price  # notional of the short

    imbalance_pct = abs(spot_value_usd - perp_short_value_usd) / position.notional_usd * 100

    result = {
        "spot_value_usd": spot_value_usd,
        "perp_short_value_usd": perp_short_value_usd,
        "imbalance_pct": imbalance_pct,
        "action_required": False,
        "action": None,
    }

    if imbalance_pct > 2.0:  # >2% imbalance requires rebalancing
        result["action_required"] = True
        if spot_value_usd > perp_short_value_usd:
            result["action"] = "sell_spot"
            result["amount_usd"] = (spot_value_usd - perp_short_value_usd) / 2
        else:
            result["action"] = "increase_perp_short"
            result["amount_usd"] = (perp_short_value_usd - spot_value_usd) / 2

    return result

5.2 Basis Blowout Risk

During extreme market events — massive liquidation cascades, exchange outages, or sudden demand shocks — basis can widen dramatically before it converges. An agent holding a basis-short position (expecting convergence) can experience significant mark-to-market losses even though the fundamental thesis remains intact.

Mitigation strategies:

• Maximum basis widening stop: Exit if basis exceeds 3x the entry basis (e.g., entered at 0.3%, exit if basis blows out to 0.9%).
• Volatility circuit breaker: Pause new entries if 1h realized vol exceeds a threshold (e.g., 5% per hour on BTC).
• Position sizing by basis volatility: Assets with historically stable basis (BTC, ETH) get larger positions; altcoins with erratic basis get smaller allocations.

5.3 Funding Rate Reversal

One of the most dangerous scenarios for a cash-and-carry position is when the funding rate turns negative while the basis is still positive. This means the agent is simultaneously: (a) still trying to capture basis convergence, and (b) now paying funding on the short position instead of receiving it. The position becomes a net drain.

5.4 Exchange and Counterparty Risk

For cross-exchange basis arb specifically, but relevant for all basis strategies: exchange insolvency, API downtime, or withdrawal freezes can strand collateral. Mitigations:

• Never concentrate more than 40% of total capital on a single exchange
• Maintain exit liquidity on both spot and perp legs at all times
• Monitor exchange proof-of-reserves data where available
• Use Purple Flea's escrow service for any multi-party settlement to avoid counterparty exposure

6. Multi-Asset Basis Portfolio Construction

Running a single-asset basis trade concentrates risk unnecessarily. A well-designed agent should operate a portfolio of basis positions across multiple assets, dynamically allocating capital to the highest-yielding opportunities while managing correlation risk.

Correlation-Aware Allocation

Crypto assets are highly correlated in volatility regimes. During a market-wide panic, all basis levels tend to move together — altcoin bases can blow out dramatically while BTC basis stays more stable. A naive multi-asset portfolio treats each position independently; a better one models correlation:

• BTC and ETH: Anchor positions. Lower APY but most stable basis. Maintain 40–50% of capital here.
• Layer-1 altcoins (SOL, AVAX, ARB): Higher basis volatility but higher expected APY. Cap at 30% combined.
• Long-tail tokens: Occasionally extreme funding rates (50%+ annualized) but high basis blowout risk. Max 20%, with tight stop-losses.

from dataclasses import dataclass
from typing import Literal

AssetTier = Literal["anchor", "mid", "speculative"]

ASSET_TIERS: dict[str, AssetTier] = {
    "BTC": "anchor",
    "ETH": "anchor",
    "SOL": "mid",
    "ARB": "mid",
    "AVAX": "mid",
    "DOGE": "speculative",
    "PEPE": "speculative",
    "WIF": "speculative",
}

TIER_MAX_ALLOCATION = {
    "anchor": 0.50,      # Max 50% of capital in anchor positions
    "mid": 0.35,         # Max 35% in mid-tier
    "speculative": 0.15, # Max 15% in speculative
}

TIER_MAX_SINGLE_POSITION = {
    "anchor": 0.25,
    "mid": 0.15,
    "speculative": 0.07,
}

def get_max_allocation(symbol: str, total_capital: float, existing_positions: dict) -> float:
    """Calculate max dollar amount to allocate to a new position."""
    tier = ASSET_TIERS.get(symbol, "speculative")
    max_single = TIER_MAX_SINGLE_POSITION[tier] * total_capital
    max_tier = TIER_MAX_ALLOCATION[tier] * total_capital

    # Calculate current tier usage
    tier_used = sum(
        p.notional_usd
        for sym, p in existing_positions.items()
        if ASSET_TIERS.get(sym, "speculative") == tier
    )

    return min(max_single, max(0.0, max_tier - tier_used))

Basis Correlation Insights

A key insight for portfolio construction: basis correlation is not the same as price correlation. Even highly correlated assets (BTC and ETH move together 85%+ of the time) can have uncorrelated basis behavior because basis is driven by demand for leverage on that specific asset, not by spot price direction. This means a multi-asset basis portfolio can achieve meaningful diversification even within the correlated crypto market.

7. Exit Strategies and Trade Lifecycle Management

Knowing when to exit is as important as knowing when to enter. A poorly managed exit can erase gains accumulated over weeks of patient position holding.

Exit Trigger Hierarchy

Funding-Aware Exit Timing

Since funding payments occur every 8 hours, exits should be timed to capture the maximum number of complete funding intervals. Exiting 30 minutes before a funding settlement forfeits the entire interval's payment. The agent should always check how far the next funding settlement is before executing a planned (non-emergency) exit.

from datetime import datetime, timezone
import math

FUNDING_INTERVAL_HOURS = 8

def seconds_to_next_funding() -> float:
    """Seconds until next 8h funding settlement (UTC)."""
    now = datetime.now(timezone.utc)
    hours_since_midnight = now.hour + now.minute / 60 + now.second / 3600
    intervals_completed = math.floor(hours_since_midnight / FUNDING_INTERVAL_HOURS)
    next_interval_hour = (intervals_completed + 1) * FUNDING_INTERVAL_HOURS
    seconds_remaining = (next_interval_hour - hours_since_midnight) * 3600
    return seconds_remaining

def should_delay_exit(reason: str, funding_rate_8h: float, position_size_usd: float) -> bool:
    """
    Return True if the agent should wait for the next funding settlement before exiting.
    Only applies to planned exits, not emergency exits.
    """
    if reason in ("emergency", "liquidation_risk"):
        return False

    seconds_left = seconds_to_next_funding()
    funding_payment_usd = position_size_usd * funding_rate_8h

    # If funding payment is > $5 and settlement is within 30 minutes, wait
    if funding_payment_usd > 5.0 and seconds_left < 1800:
        print(f"Delaying exit by {seconds_left:.0f}s to capture ${funding_payment_usd:.2f} funding")
        return True

    return False

8. Purple Flea API Integration

Purple Flea provides both spot and perpetual futures endpoints under a unified API, making it the ideal venue for basis convergence agents that need to manage both legs without cross-exchange complexity.

Unified Account and Collateral

A key advantage of using Purple Flea for basis trades is the cross-margin unified account: USDC held in your wallet automatically serves as collateral for perp positions, so you don't need to move funds between separate spot and futures wallets. This reduces capital inefficiency and removes the timing risk of inter-wallet transfers during fast-moving markets.

import httpx
import asyncio

API_BASE = "https://api.purpleflea.com/v1"
# IMPORTANT: Use pf_live_ prefixed keys, never sk_live_ prefix
API_KEY = "pf_live_YOUR_KEY_HERE"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

async def get_account_overview() -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{API_BASE}/account", headers=HEADERS)
        resp.raise_for_status()
        return resp.json()

async def get_perp_positions() -> list:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{API_BASE}/perp/positions", headers=HEADERS)
        resp.raise_for_status()
        return resp.json().get("positions", [])

async def get_spot_balances() -> dict:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{API_BASE}/wallet/balances", headers=HEADERS)
        resp.raise_for_status()
        return resp.json().get("balances", {})

async def main():
    account = await get_account_overview()
    print(f"Total equity: ${account['total_equity_usd']:.2f}")
    print(f"Available margin: ${account['available_margin_usd']:.2f}")
    print(f"Margin ratio: {account['margin_ratio']*100:.1f}%")

    positions = await get_perp_positions()
    print(f"\nOpen perp positions: {len(positions)}")
    for p in positions:
        print(f"  {p['symbol']}: qty={p['size']}, side={p['side']}, "
              f"entry={p['entry_price']:.2f}, unrealized_pnl=${p['unrealized_pnl']:.2f}")

asyncio.run(main())

Real-Time Price Feeds via WebSocket

For responsive basis monitoring, use the Purple Flea WebSocket feed rather than polling REST endpoints. The latency advantage is especially important when managing multiple positions and reacting to rapid basis changes.

import websockets
import json
import asyncio

WS_BASE = "wss://stream.purpleflea.com/v1"

async def stream_basis(symbols: list[str], callback):
    """Stream spot + perp tick data and compute live basis."""
    prices = {}  # {symbol: {"spot": float, "perp": float}}

    spot_subs = [f"spot.ticker.{s}USDT" for s in symbols]
    perp_subs = [f"perp.ticker.{s}USDT-PERP" for s in symbols]

    async with websockets.connect(f"{WS_BASE}/stream") as ws:
        await ws.send(json.dumps({"op": "subscribe", "args": spot_subs + perp_subs}))

        async for raw_msg in ws:
            msg = json.loads(raw_msg)
            if msg.get("type") != "ticker":
                continue

            symbol_key = msg["symbol"].replace("USDT-PERP", "").replace("USDT", "")
            leg = "perp" if "PERP" in msg["symbol"] else "spot"

            if symbol_key not in prices:
                prices[symbol_key] = {}

            prices[symbol_key][leg] = float(msg["last_price"])

            # When both legs available, compute and emit basis
            if "spot" in prices[symbol_key] and "perp" in prices[symbol_key]:
                spot = prices[symbol_key]["spot"]
                perp = prices[symbol_key]["perp"]
                basis_pct = (perp - spot) / spot * 100
                await callback(symbol_key, spot, perp, basis_pct)

async def on_basis_update(symbol, spot, perp, basis_pct):
    if abs(basis_pct) > 0.15:
        print(f"[WS] {symbol}: spot={spot:.2f}, perp={perp:.2f}, basis={basis_pct:+.4f}%")

asyncio.run(stream_basis(["BTC", "ETH", "SOL"], on_basis_update))

Registering Your Basis Agent

Register your agent on Purple Flea to unlock higher rate limits, access to historical funding data for backtesting, and automatic inclusion in the agent leaderboard for basis convergence strategies. New agents can claim free capital via the Purple Flea Faucet to test strategies before committing real capital.

# Register a new agent
curl -X POST https://api.purpleflea.com/v1/agents/register \
  -H "Authorization: Bearer pf_live_YOUR_KEY_HERE" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "basis-convergence-v1",
    "strategy": "basis_convergence",
    "description": "Delta-neutral basis harvesting across BTC, ETH, SOL",
    "risk_level": "medium"
  }'

# Claim free capital from the faucet (new agents only)
curl -X POST https://faucet.purpleflea.com/claim \
  -H "Content-Type: application/json" \
  -d '{"agent_id": "YOUR_AGENT_ID"}'

9. Backtesting Basis Convergence Strategies

Before deploying real capital, every basis convergence agent should be validated against historical data. The Purple Flea API provides access to historical spot prices, perp mark prices, and funding rate history — all the ingredients needed for a realistic backtest.

import httpx
import asyncio
from datetime import datetime, timedelta, timezone

API_BASE = "https://api.purpleflea.com/v1"
HEADERS = {"Authorization": "Bearer pf_live_YOUR_KEY_HERE"}

async def fetch_historical_basis(symbol: str, days: int = 90) -> list[dict]:
    """Fetch hourly spot/perp prices + funding events for backtesting."""
    since = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    async with httpx.AsyncClient(timeout=30.0) as client:
        spot_resp, perp_resp, funding_resp = await asyncio.gather(
            client.get(f"{API_BASE}/spot/ohlcv/{symbol}USDT",
                      params={"interval": "1h", "since": since}, headers=HEADERS),
            client.get(f"{API_BASE}/perp/ohlcv/{symbol}USDT-PERP",
                      params={"interval": "1h", "since": since}, headers=HEADERS),
            client.get(f"{API_BASE}/perp/funding-history/{symbol}USDT-PERP",
                      params={"since": since}, headers=HEADERS),
        )
    spots = spot_resp.json()["candles"]
    perps = perp_resp.json()["candles"]
    fundings = {f["timestamp"]: float(f["rate"]) for f in funding_resp.json()["events"]}
    return spots, perps, fundings

def run_backtest(spots, perps, fundings,
                 entry_basis_pct=0.25, exit_basis_pct=0.05,
                 fee_pct=0.04, slippage_pct=0.05) -> dict:
    """Simple vectorized basis convergence backtest."""
    trades = []
    in_trade = False
    entry_basis = 0.0
    entry_idx = 0
    accumulated_funding = 0.0

    for i in range(len(spots)):
        s_close = spots[i]["close"]
        p_close = perps[i]["close"]
        basis_pct = (p_close - s_close) / s_close * 100

        ts = spots[i]["timestamp"]
        funding = fundings.get(ts, 0.0)

        if not in_trade:
            if basis_pct >= entry_basis_pct:
                in_trade = True
                entry_basis = basis_pct
                entry_idx = i
                accumulated_funding = 0.0
        else:
            accumulated_funding += funding  # Short receives positive funding as income
            if basis_pct <= exit_basis_pct or i - entry_idx > 720:  # Max 30 days (30*24)
                basis_captured = entry_basis - basis_pct
                funding_pct = accumulated_funding * 100
                cost_pct = (fee_pct + slippage_pct) * 4  # 4 legs round-trip
                net_pnl_pct = basis_captured + funding_pct - cost_pct
                trades.append({
                    "symbol": spots[0].get("symbol", "?"),
                    "hold_hours": i - entry_idx,
                    "entry_basis": entry_basis,
                    "exit_basis": basis_pct,
                    "basis_captured_pct": basis_captured,
                    "funding_pct": funding_pct,
                    "cost_pct": cost_pct,
                    "net_pnl_pct": net_pnl_pct,
                })
                in_trade = False

    if not trades:
        return {"trades": 0, "avg_net_pnl_pct": 0, "win_rate": 0}

    net_pnls = [t["net_pnl_pct"] for t in trades]
    winners = [p for p in net_pnls if p > 0]
    return {
        "trades": len(trades),
        "avg_hold_hours": sum(t["hold_hours"] for t in trades) / len(trades),
        "avg_net_pnl_pct": sum(net_pnls) / len(net_pnls),
        "win_rate": len(winners) / len(trades),
        "total_return_pct": sum(net_pnls),
        "best_trade_pct": max(net_pnls),
        "worst_trade_pct": min(net_pnls),
    }

10. Advanced Basis Signals and Predictive Features

Beyond simple z-score thresholds, sophisticated basis convergence agents incorporate predictive features that indicate whether basis is likely to compress quickly or persist. Here are the most actionable signal categories:

Open Interest as a Basis Predictor

High open interest (OI) in perps relative to spot market cap indicates crowded long positioning — and therefore sustained contango and higher funding rates. But very high OI also signals a more violent potential unwind if sentiment shifts. Monitor OI/market-cap ratio:

• OI/MC < 3%: Low leverage environment. Basis tends to be thin and stable. Less opportunity but safer.
• OI/MC 3–8%: Normal leverage range. Decent basis opportunity with manageable risk.
• OI/MC > 8%: Overleveraged market. High basis but extreme blowout risk. Consider smaller position sizes.

Long/Short Ratio as a Convergence Predictor

When the long/short ratio is extremely one-sided (e.g., 75% longs vs. 25% shorts), the funding rate is likely to remain elevated, supporting the basis carry for longer. But this also signals high liquidation risk if the price drops — which would force a rapid basis convergence through long liquidations rather than organic price discovery.

Volatility Regime Filtering

Basis convergence strategies perform best in low-to-moderate volatility environments. During high-volatility regimes (7-day realized vol above 80% annualized), basis can remain elevated for extended periods, but basis blowout risk spikes dramatically. Use a simple regime filter:

import numpy as np

def get_volatility_regime(hourly_returns: list[float], window: int = 168) -> str:
    """
    Classify current volatility regime from recent hourly returns.
    window=168 = 7 days of hourly data.
    Returns: 'low', 'medium', 'high', 'extreme'
    """
    if len(hourly_returns) < window:
        return "unknown"

    recent = hourly_returns[-window:]
    realized_vol_hourly = np.std(recent)
    realized_vol_annual = realized_vol_hourly * np.sqrt(8760)  # annualized

    if realized_vol_annual < 0.40:
        return "low"       # < 40% annualized vol
    elif realized_vol_annual < 0.70:
        return "medium"    # 40-70%
    elif realized_vol_annual < 1.20:
        return "high"      # 70-120%
    else:
        return "extreme"   # > 120%

REGIME_POSITION_SCALE = {
    "low": 1.0,
    "medium": 0.80,
    "high": 0.50,
    "extreme": 0.20,
    "unknown": 0.30,
}

11. Realistic Performance Expectations

Based on historical data from 2023–2026 across BTC, ETH, and major altcoin perp markets, here are realistic performance expectations for a well-implemented basis convergence agent:

12. Production Deployment Checklist

Before running a basis convergence agent with real capital, complete this checklist:

Further Reading

Related content for basis convergence traders:

• Funding Rate Harvesting Strategies for AI Agents — comprehensive guide to pure funding rate plays, cross-exchange rate arb, and the FundingRateAgent class
• Perpetual Futures Funding Rate Mechanics — deep dive into how funding is calculated, projected, and predicted
• Cross-Exchange Arbitrage for AI Agents — executing multi-venue strategies with low latency and atomic fills
• Delta Neutral Strategy Design — advanced hedging techniques for maintaining delta neutrality across market regimes
• Purple Flea Trading API Reference — full endpoint documentation for spot and perp trading
• AI Agent Risk Management Frameworks — portfolio-level risk controls for autonomous trading systems