Market-Neutral Strategies for AI Trading Agents: Zero Beta, Pure Alpha

The most sophisticated AI trading agents in 2026 do not bet on market direction — they extract pure alpha while maintaining zero net exposure to systematic risk. Market-neutral strategies let agents profit in bull markets, bear markets, and sideways drift alike. This guide covers the mathematics, the code, and the Purple Flea API integration needed to build a production market-neutral agent.

What Market-Neutral Actually Means

A market-neutral portfolio is one where the expected return is independent of the market's return. Formally, if R_p is the portfolio return and R_m is the market return, then:

The beta (β) of a portfolio measures its sensitivity to market movements. A beta of 1.0 means the portfolio moves in lockstep with the market. A beta of 0 means market moves have no effect — the agent's returns are driven entirely by its edge, not by whether the S&P 500 is up or down today.

Types of Neutrality

Market-neutral is a broad umbrella. Agents must choose which risks they neutralise and which they retain:

Why AI Agents Particularly Benefit

Human hedge fund managers maintain market-neutral books with enormous staffs, risk systems, and compliance overhead. AI agents have structural advantages:

• Continuous recalibration: Agents can recompute beta exposures every tick and rebalance instantly without fatigue or hesitation.
• No career risk bias: Human PMs drift toward long bias in bull markets to avoid underperformance. Agents optimise objectively.
• Parallelism: A single agent can simultaneously manage 50+ spread positions, each requiring independent hedge calculations.
• Speed: Hedge adjustments that would take a human team hours can be executed by an agent in milliseconds via the Purple Flea trading API.

Long-Short Equity: Factor-Neutral Construction

Long-short equity is the most common market-neutral strategy. The agent goes long stocks it expects to outperform and short stocks it expects to underperform, with the hedge ratio set to achieve zero net beta.

Factor Exposure in Long-Short Books

A naive long-short that is merely dollar-neutral still carries enormous factor risk. To be truly neutral, agents must model and neutralise exposures across multiple systematic factors:

In practice, agents pull factor loadings from a risk model (Barra, Axioma, or a custom model built from historical returns), then solve a quadratic program to find the optimal set of weights that maximises expected alpha subject to near-zero factor exposures and position limits.

Portfolio Construction Algorithm

Python
import numpy as np
from scipy.optimize import minimize

def construct_factor_neutral_portfolio(
    alpha_scores: dict,      # {ticker: expected_alpha_score}
    factor_exposures: dict,  # {ticker: {factor: exposure}}
    beta_estimates: dict,    # {ticker: market_beta}
    max_position: float = 0.05,  # max 5% per position
    max_sector_tilt: float = 0.10
) -> dict:
    """
    Build a factor-neutral long-short portfolio maximising alpha.
    Returns weights dict: positive = long, negative = short.
    """
    tickers = list(alpha_scores.keys())
    n = len(tickers)
    alphas = np.array([alpha_scores[t] for t in tickers])

    # Factor exposure matrix [n_stocks x n_factors]
    factors = ['market', 'value', 'momentum', 'size', 'quality']
    F = np.array([
        [factor_exposures[t].get(f, 0.0) for f in factors]
        for t in tickers
    ])

    def objective(weights):
        # Maximise alpha (minimise negative alpha)
        return -np.dot(weights, alphas)

    def gradient(weights):
        return -alphas

    constraints = []

    # Beta-neutral: portfolio beta = 0
    betas = np.array([beta_estimates[t] for t in tickers])
    constraints.append({
        'type': 'eq',
        'fun': lambda w: np.dot(w, betas),
        'jac': lambda w: betas
    })

    # Dollar-neutral: sum of weights = 0
    constraints.append({
        'type': 'eq',
        'fun': lambda w: np.sum(w),
        'jac': lambda w: np.ones(n)
    })

    # Factor neutral: each factor exposure near zero
    for fi, factor in enumerate(factors[1:], 1):  # skip market (already in beta constraint)
        fi_idx = fi
        constraints.append({
            'type': 'ineq',
            'fun': lambda w, idx=fi_idx: 0.02 - abs(np.dot(w, F[:, idx])),
        })

    # Gross leverage = 2x (sum of |weights| = 2)
    constraints.append({
        'type': 'eq',
        'fun': lambda w: np.sum(np.abs(w)) - 2.0
    })

    bounds = [(-max_position, max_position)] * n
    w0 = np.zeros(n)
    w0[:n//2] = 2.0 / n
    w0[n//2:] = -2.0 / n

    result = minimize(
        objective,
        w0,
        method='SLSQP',
        jac=gradient,
        bounds=bounds,
        constraints=constraints,
        options={'maxiter': 1000, 'ftol': 1e-9}
    )

    if not result.success:
        raise RuntimeError(f"Portfolio optimisation failed: {result.message}")

    return {tickers[i]: result.x[i] for i in range(n) if abs(result.x[i]) > 0.001}

Statistical Arbitrage: Mean-Reverting Spread Trading

Statistical arbitrage (stat arb) exploits temporary divergences between correlated assets. Unlike fundamental long-short, stat arb relies on the statistical property that correlated assets' spread (price ratio or log-price difference) mean-reverts over time.

Cointegration and Spread Formation

Two price series P_A and P_B are cointegrated if a linear combination is stationary. The classic test is the Engle-Granger two-step:

Entry and Exit Signals

Stat Arb Implementation

Python
import numpy as np
from statsmodels.tsa.stattools import adfuller, coint
from statsmodels.regression.linear_model import OLS
from statsmodels.tools import add_constant
from collections import deque
from dataclasses import dataclass
from typing import Optional
import asyncio

@dataclass
class SpreadPosition:
    pair: tuple[str, str]
    hedge_ratio: float
    entry_zscore: float
    direction: int  # +1 = long spread, -1 = short spread
    long_ticker: str
    short_ticker: str
    size_usd: float

class StatArbAgent:
    """
    Statistical arbitrage agent with cointegration-based pair selection
    and real-time Z-score signal generation.
    """

    def __init__(
        self,
        lookback: int = 252,         # trading days for calibration
        zscore_entry: float = 2.0,   # entry threshold
        zscore_exit: float = 0.5,    # profit-taking threshold
        zscore_stop: float = 3.5,    # stop-loss threshold
        max_pairs: int = 20,
        position_size_usd: float = 10_000,
    ):
        self.lookback = lookback
        self.zscore_entry = zscore_entry
        self.zscore_exit = zscore_exit
        self.zscore_stop = zscore_stop
        self.max_pairs = max_pairs
        self.position_size_usd = position_size_usd

        self.price_history: dict[str, deque] = {}
        self.cointegrated_pairs: dict[tuple, dict] = {}
        self.active_positions: dict[tuple, SpreadPosition] = {}

    def update_price(self, ticker: str, price: float):
        if ticker not in self.price_history:
            self.price_history[ticker] = deque(maxlen=self.lookback)
        self.price_history[ticker].append(price)

    def find_cointegrated_pairs(self, universe: list[str]) -> list[tuple]:
        """
        Scan universe for cointegrated pairs using Engle-Granger test.
        Returns pairs sorted by ADF t-statistic (most stationary first).
        """
        eligible = [t for t in universe if len(self.price_history.get(t, [])) >= self.lookback]
        valid_pairs = []

        for i, t1 in enumerate(eligible):
            for t2 in eligible[i+1:]:
                p1 = np.log(np.array(self.price_history[t1]))
                p2 = np.log(np.array(self.price_history[t2]))

                # Engle-Granger cointegration test
                score, pvalue, _ = coint(p1, p2)
                if pvalue < 0.05:
                    # Compute hedge ratio via OLS
                    gamma, resid = self._compute_hedge_ratio(p1, p2)
                    adf_stat, adf_p, *_ = adfuller(resid, maxlag=1)
                    half_life = self._half_life(resid)

                    if 5 < half_life < 120:  # practical mean-reversion window
                        valid_pairs.append({
                            'pair': (t1, t2),
                            'hedge_ratio': gamma,
                            'adf_stat': adf_stat,
                            'adf_pvalue': adf_p,
                            'coint_pvalue': pvalue,
                            'half_life_days': half_life,
                            'spread_std': np.std(resid),
                            'spread_mean': np.mean(resid),
                        })

        # Sort by ADF statistic (more negative = more stationary = better)
        valid_pairs.sort(key=lambda x: x['adf_stat'])
        selected = valid_pairs[:self.max_pairs]

        # Cache calibration data
        for p in selected:
            self.cointegrated_pairs[p['pair']] = p

        return [p['pair'] for p in selected]

    def _compute_hedge_ratio(self, log_p1: np.ndarray, log_p2: np.ndarray):
        X = add_constant(log_p2)
        model = OLS(log_p1, X).fit()
        gamma = model.params[1]
        resid = log_p1 - gamma * log_p2 - model.params[0]
        return gamma, resid

    def _half_life(self, spread: np.ndarray) -> float:
        """Ornstein-Uhlenbeck half-life estimation."""
        lag = spread[:-1]
        delta = spread[1:] - spread[:-1]
        X = add_constant(lag)
        model = OLS(delta, X).fit()
        kappa = -model.params[1]
        return np.log(2) / kappa if kappa > 0 else np.inf

    def compute_zscore(self, pair: tuple) -> Optional[float]:
        if pair not in self.cointegrated_pairs:
            return None
        meta = self.cointegrated_pairs[pair]
        t1, t2 = pair

        if not (self.price_history.get(t1) and self.price_history.get(t2)):
            return None

        p1 = np.log(self.price_history[t1][-1])
        p2 = np.log(self.price_history[t2][-1])
        spread = p1 - meta['hedge_ratio'] * p2
        zscore = (spread - meta['spread_mean']) / meta['spread_std']
        return zscore

    def generate_signals(self) -> list[dict]:
        signals = []

        for pair, meta in self.cointegrated_pairs.items():
            zscore = self.compute_zscore(pair)
            if zscore is None:
                continue
            t1, t2 = pair

            if pair in self.active_positions:
                pos = self.active_positions[pair]
                # Check exit conditions
                if abs(zscore) < self.zscore_exit:
                    signals.append({'action': 'close', 'pair': pair, 'reason': 'mean_reverted', 'zscore': zscore})
                elif abs(zscore) > self.zscore_stop:
                    signals.append({'action': 'close', 'pair': pair, 'reason': 'stop_loss', 'zscore': zscore})
            else:
                # Check entry conditions
                if zscore < -self.zscore_entry:
                    signals.append({
                        'action': 'open',
                        'pair': pair,
                        'direction': 1,   # long spread
                        'long': t1,
                        'short': t2,
                        'hedge_ratio': meta['hedge_ratio'],
                        'zscore': zscore,
                        'half_life': meta['half_life_days'],
                        'size_usd': self.position_size_usd
                    })
                elif zscore > self.zscore_entry:
                    signals.append({
                        'action': 'open',
                        'pair': pair,
                        'direction': -1,  # short spread
                        'long': t2,
                        'short': t1,
                        'hedge_ratio': 1.0 / meta['hedge_ratio'],
                        'zscore': zscore,
                        'half_life': meta['half_life_days'],
                        'size_usd': self.position_size_usd
                    })

        return signals

Volatility-Neutral: Delta-Hedged Options Positions

Volatility-neutral strategies take positions in options to express views on implied vs. realised volatility, while hedging out the directional price risk with the underlying asset or futures.

Delta Hedging Mechanics

An option's delta measures how much the option price moves per unit move in the underlying. To achieve a delta-neutral position, the agent holds offsetting underlying positions:

Common Volatility-Neutral Structures

Python — Delta Hedger
import math
from dataclasses import dataclass, field
from typing import Literal
import numpy as np

@dataclass
class OptionPosition:
    ticker: str
    strike: float
    expiry_days: float
    option_type: Literal['call', 'put']
    quantity: int        # positive = long, negative = short
    implied_vol: float
    underlying_price: float

    def _d1(self, S: float, sigma: float, T: float, r: float = 0.05) -> float:
        return (math.log(S / self.strike) + (r + 0.5 * sigma**2) * T) / (sigma * math.sqrt(T))

    def delta(self, S: float, sigma: float, T: float, r: float = 0.05) -> float:
        from scipy.stats import norm
        d1 = self._d1(S, sigma, T, r)
        if self.option_type == 'call':
            return norm.cdf(d1)
        else:
            return norm.cdf(d1) - 1.0

    def gamma(self, S: float, sigma: float, T: float, r: float = 0.05) -> float:
        from scipy.stats import norm
        d1 = self._d1(S, sigma, T, r)
        return norm.pdf(d1) / (S * sigma * math.sqrt(T))

    def vega(self, S: float, sigma: float, T: float, r: float = 0.05) -> float:
        from scipy.stats import norm
        d1 = self._d1(S, sigma, T, r)
        return S * norm.pdf(d1) * math.sqrt(T) / 100  # per 1% vol move


class DeltaNeutralVolPortfolio:
    """
    Manages a portfolio of options with continuous delta hedging.
    Goal: zero delta exposure, long/short vega for vol trades.
    """

    def __init__(self, hedge_frequency_minutes: int = 5):
        self.options: list[OptionPosition] = []
        self.underlying_hedge: dict[str, float] = {}  # ticker → shares held for hedge
        self.hedge_freq = hedge_frequency_minutes

    def add_position(self, opt: OptionPosition):
        self.options.append(opt)

    def portfolio_greeks(self, current_prices: dict[str, float]) -> dict:
        total_delta = 0.0
        total_gamma = 0.0
        total_vega = 0.0
        total_theta = 0.0

        for opt in self.options:
            S = current_prices.get(opt.ticker, opt.underlying_price)
            T = opt.expiry_days / 365.0
            if T <= 0:
                continue
            multiplier = opt.quantity * 100  # standard option = 100 shares

            total_delta += opt.delta(S, opt.implied_vol, T) * multiplier
            total_gamma += opt.gamma(S, opt.implied_vol, T) * multiplier
            total_vega  += opt.vega(S, opt.implied_vol, T) * multiplier

        # Add hedge delta
        for ticker, shares in self.underlying_hedge.items():
            total_delta += shares

        return {
            'net_delta': round(total_delta, 4),
            'net_gamma': round(total_gamma, 6),
            'net_vega': round(total_vega, 4),
            'is_delta_neutral': abs(total_delta) < 0.5,
        }

    def compute_hedge_adjustments(self, current_prices: dict[str, float]) -> dict[str, float]:
        """
        Returns required share adjustments to achieve delta neutrality
        for each underlying ticker.
        """
        option_deltas: dict[str, float] = {}

        for opt in self.options:
            S = current_prices.get(opt.ticker, opt.underlying_price)
            T = opt.expiry_days / 365.0
            if T <= 0:
                continue
            delta_shares = opt.delta(S, opt.implied_vol, T) * opt.quantity * 100
            option_deltas[opt.ticker] = option_deltas.get(opt.ticker, 0.0) + delta_shares

        adjustments = {}
        for ticker, option_delta in option_deltas.items():
            current_hedge = self.underlying_hedge.get(ticker, 0.0)
            required_hedge = -option_delta
            adjustment = required_hedge - current_hedge
            if abs(adjustment) > 0.5:  # minimum trade threshold
                adjustments[ticker] = round(adjustment, 2)
                self.underlying_hedge[ticker] = required_hedge

        return adjustments

The MarketNeutralAgent Class

The following is a complete, production-ready MarketNeutralAgent class that integrates beta calculation, hedge ratio computation, and portfolio construction into a unified agent architecture designed to work with the Purple Flea trading API.

Python — MarketNeutralAgent (full implementation)
import asyncio
import numpy as np
import httpx
from dataclasses import dataclass, field
from typing import Optional
from collections import defaultdict, deque

PF_API_BASE = "https://api.purpleflea.com/v1"
PF_API_KEY  = "pf_live_your_api_key_here"

@dataclass
class Position:
    ticker: str
    quantity: float   # positive = long, negative = short
    entry_price: float
    current_price: float = 0.0
    beta: float = 1.0

    @property
    def market_value(self) -> float:
        return self.quantity * self.current_price

    @property
    def beta_adjusted_value(self) -> float:
        return self.market_value * self.beta

    @property
    def pnl(self) -> float:
        return self.quantity * (self.current_price - self.entry_price)


class MarketNeutralAgent:
    """
    A full-featured market-neutral trading agent.
    Maintains zero net beta exposure while harvesting idiosyncratic alpha.
    """

    def __init__(
        self,
        api_key: str = PF_API_KEY,
        market_ticker: str = "SPY",          # benchmark for beta calculation
        beta_window: int = 63,               # 63 trading days = ~3 months
        rebalance_threshold: float = 0.10,   # rebalance if |beta| > 0.10
        max_gross_leverage: float = 2.0,
    ):
        self.api_key = api_key
        self.market_ticker = market_ticker
        self.beta_window = beta_window
        self.rebalance_threshold = rebalance_threshold
        self.max_gross_leverage = max_gross_leverage

        self.positions: dict[str, Position] = {}
        self.price_history: dict[str, deque] = defaultdict(lambda: deque(maxlen=beta_window + 1))
        self.market_history: deque = deque(maxlen=beta_window + 1)
        self.client = httpx.AsyncClient(
            base_url=PF_API_BASE,
            headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
            timeout=10.0
        )

    # ─── Beta Calculation ────────────────────────────────────────────────────

    def calculate_beta(self, ticker: str) -> Optional[float]:
        """
        Compute OLS beta of ticker returns against market returns.
        Returns None if insufficient history.
        """
        stock_prices = list(self.price_history[ticker])
        market_prices = list(self.market_history)

        min_len = min(len(stock_prices), len(market_prices))
        if min_len < 20:
            return None  # insufficient data

        stock_prices = np.array(stock_prices[-min_len:])
        market_prices = np.array(market_prices[-min_len:])

        stock_returns = np.diff(np.log(stock_prices))
        market_returns = np.diff(np.log(market_prices))

        if len(stock_returns) < 10:
            return None

        # OLS: stock_returns = alpha + beta * market_returns + epsilon
        cov_matrix = np.cov(stock_returns, market_returns)
        beta = cov_matrix[0, 1] / cov_matrix[1, 1]
        return float(beta)

    def calculate_portfolio_beta(self) -> float:
        """
        Compute dollar-weighted portfolio beta across all positions.
        Beta-neutral target: this should be 0.
        """
        total_market_value = sum(abs(p.market_value) for p in self.positions.values())
        if total_market_value == 0:
            return 0.0

        weighted_beta = 0.0
        for pos in self.positions.values():
            beta = self.calculate_beta(pos.ticker) or 1.0
            pos.beta = beta
            # Weight by dollar value with sign (long positive, short negative)
            weight = pos.market_value / total_market_value
            weighted_beta += weight * beta

        return weighted_beta

    # ─── Hedge Ratio ─────────────────────────────────────────────────────────

    def hedge_ratio(
        self,
        long_ticker: str,
        short_ticker: str
    ) -> float:
        """
        Compute hedge ratio: how many units of short_ticker to short
        per unit of long_ticker, to achieve zero net beta contribution.

        hedge_ratio = (beta_long * price_long) / (beta_short * price_short)
        """
        beta_long  = self.calculate_beta(long_ticker)  or 1.0
        beta_short = self.calculate_beta(short_ticker) or 1.0

        price_long  = list(self.price_history[long_ticker])[-1]  if self.price_history[long_ticker]  else 1.0
        price_short = list(self.price_history[short_ticker])[-1] if self.price_history[short_ticker] else 1.0

        if beta_short * price_short == 0:
            return 1.0

        ratio = (beta_long * price_long) / (beta_short * price_short)
        return round(ratio, 4)

    # ─── Portfolio Construction ───────────────────────────────────────────────

    def construct_portfolio(
        self,
        long_candidates: list[dict],   # [{'ticker': str, 'alpha_score': float}, ...]
        short_candidates: list[dict],
        total_capital: float,
        n_pairs: int = 10
    ) -> list[dict]:
        """
        Construct a market-neutral portfolio from alpha-ranked candidates.
        Returns a list of orders to execute.

        Strategy: pair highest-alpha longs with lowest-alpha shorts,
        sizing each pair to achieve beta-neutral contribution.
        """
        longs  = sorted(long_candidates,  key=lambda x: -x['alpha_score'])[:n_pairs]
        shorts = sorted(short_candidates, key=lambda x:  x['alpha_score'])[:n_pairs]

        capital_per_pair = total_capital / n_pairs
        orders = []

        for i in range(min(len(longs), len(shorts))):
            long_t  = longs[i]['ticker']
            short_t = shorts[i]['ticker']

            beta_l = self.calculate_beta(long_t)  or 1.0
            beta_s = self.calculate_beta(short_t) or 1.0

            price_l = list(self.price_history[long_t])[-1]  if self.price_history[long_t]  else None
            price_s = list(self.price_history[short_t])[-1] if self.price_history[short_t] else None

            if price_l is None or price_s is None:
                continue

            # Dollar-size longs and shorts to be beta-equivalent
            # Long notional: capital_per_pair / 2
            long_notional = capital_per_pair / 2

            # Short notional adjusted for beta ratio
            beta_ratio = beta_l / beta_s if beta_s != 0 else 1.0
            short_notional = long_notional * beta_ratio

            long_qty  = int(long_notional  / price_l)
            short_qty = int(short_notional / price_s)

            if long_qty > 0 and short_qty > 0:
                orders.append({
                    'ticker': long_t,
                    'side': 'buy',
                    'quantity': long_qty,
                    'order_type': 'market',
                    'pair_id': f"pair_{i}",
                    'strategy': 'market_neutral_long'
                })
                orders.append({
                    'ticker': short_t,
                    'side': 'sell',
                    'quantity': short_qty,
                    'order_type': 'market',
                    'pair_id': f"pair_{i}",
                    'strategy': 'market_neutral_short'
                })

        return orders

    # ─── Monitoring & Rebalancing ─────────────────────────────────────────────

    def needs_rebalance(self) -> tuple[bool, float]:
        """Check if portfolio beta drift exceeds threshold."""
        current_beta = self.calculate_portfolio_beta()
        return abs(current_beta) > self.rebalance_threshold, current_beta

    async def get_beta_hedge(self, target_beta_delta: float, portfolio_value: float) -> dict:
        """
        Compute futures hedge order to restore beta neutrality.
        Uses SPY futures as the hedge instrument.

        target_beta_delta: how much beta to offset (e.g., +0.15 → short 0.15 beta-equivalent)
        """
        spy_price = list(self.market_history)[-1] if self.market_history else 500.0
        spy_futures_multiplier = 50  # ES mini = $50 per point

        # Dollar value to hedge
        hedge_value = target_beta_delta * portfolio_value
        # Number of futures contracts
        n_contracts = hedge_value / (spy_price * spy_futures_multiplier)
        n_contracts_rounded = round(n_contracts)

        return {
            'instrument': 'ES',   # S&P 500 e-mini futures
            'side': 'sell' if n_contracts_rounded > 0 else 'buy',
            'quantity': abs(n_contracts_rounded),
            'order_type': 'market',
            'purpose': 'beta_hedge',
            'target_beta_offset': -target_beta_delta
        }

    async def update_prices(self, prices: dict[str, float]):
        """Ingest new price tick for all tracked assets."""
        for ticker, price in prices.items():
            self.price_history[ticker].append(price)
            if ticker in self.positions:
                self.positions[ticker].current_price = price
        if self.market_ticker in prices:
            self.market_history.append(prices[self.market_ticker])

    def portfolio_summary(self) -> dict:
        """Return current portfolio state and risk metrics."""
        longs  = [p for p in self.positions.values() if p.quantity > 0]
        shorts = [p for p in self.positions.values() if p.quantity < 0]

        gross_exposure = sum(abs(p.market_value) for p in self.positions.values())
        net_exposure   = sum(p.market_value for p in self.positions.values())
        total_pnl      = sum(p.pnl for p in self.positions.values())
        portfolio_beta = self.calculate_portfolio_beta()

        return {
            'n_longs': len(longs),
            'n_shorts': len(shorts),
            'gross_exposure': round(gross_exposure, 2),
            'net_exposure': round(net_exposure, 2),
            'net_dollar_neutral': abs(net_exposure) < gross_exposure * 0.05,
            'portfolio_beta': round(portfolio_beta, 4),
            'beta_neutral': abs(portfolio_beta) < self.rebalance_threshold,
            'total_pnl': round(total_pnl, 2),
            'long_pnl':  round(sum(p.pnl for p in longs), 2),
            'short_pnl': round(sum(p.pnl for p in shorts), 2),
        }

Beta Hedging with Futures Contracts

When a portfolio accumulates beta drift — either from position-level changes, price moves, or new additions — futures contracts provide the fastest and cheapest instrument for restoring neutrality without touching existing stock positions.

Futures Hedge Mechanics

Common futures instruments for equity beta hedging:

Measuring Neutrality: Ex-Ante vs Ex-Post Beta

The gap between predicted neutrality (ex-ante) and realised neutrality (ex-post) is where most market-neutral strategies leak alpha back to the market. Understanding and minimising this gap is critical.

Ex-Ante Beta

Ex-ante beta is the expected portfolio beta computed before realised returns. It is based on the covariance model used during portfolio construction — typically a risk model with 3–5 years of historical data, factor loadings, and an assumed covariance structure.

Ex-Post Beta

Ex-post beta is measured from realised portfolio returns. After the fact, regress daily portfolio returns against market returns:

Python — Ex-Post Beta Analysis
import numpy as np
from scipy import stats

def analyse_realised_neutrality(
    portfolio_returns: list[float],   # daily portfolio return series
    market_returns: list[float],      # daily market (SPY) return series
    confidence: float = 0.95
) -> dict:
    """
    Compute ex-post beta and statistical significance.
    Assess whether portfolio achieved true market neutrality.
    """
    port_r = np.array(portfolio_returns)
    mkt_r  = np.array(market_returns)
    n = len(port_r)

    # OLS regression: R_port = alpha + beta * R_market + epsilon
    slope, intercept, r_value, p_value, std_err = stats.linregress(mkt_r, port_r)

    beta_expost  = slope
    alpha_annual = intercept * 252  # annualise daily alpha
    r_squared    = r_value ** 2

    # t-statistic for beta = 0 hypothesis
    t_stat_beta = beta_expost / std_err
    # p-value: is beta statistically different from zero?
    p_beta_zero = 2 * (1 - stats.t.cdf(abs(t_stat_beta), df=n-2))

    # Confidence interval for beta
    t_crit = stats.t.ppf((1 + confidence) / 2, df=n-2)
    beta_ci_low  = beta_expost - t_crit * std_err
    beta_ci_high = beta_expost + t_crit * std_err

    # Residual analysis (idiosyncratic return)
    fitted = intercept + slope * mkt_r
    residuals = port_r - fitted
    tracking_error = np.std(residuals) * np.sqrt(252)
    information_ratio = (alpha_annual / tracking_error) if tracking_error > 0 else 0

    # Sharpe ratio (using risk-free = 5%)
    excess_returns = port_r - 0.05 / 252
    sharpe = np.mean(excess_returns) / np.std(excess_returns) * np.sqrt(252)

    # Beta drift analysis — rolling 21-day beta
    rolling_betas = []
    for i in range(21, n):
        window_port = port_r[i-21:i]
        window_mkt  = mkt_r[i-21:i]
        b, *_ = stats.linregress(window_mkt, window_port)
        rolling_betas.append(b)

    return {
        'ex_post_beta': round(beta_expost, 4),
        'beta_p_value': round(p_beta_zero, 4),
        'beta_significant': p_beta_zero < (1 - confidence),
        'beta_ci_95': (round(beta_ci_low, 4), round(beta_ci_high, 4)),
        'truly_neutral': abs(beta_expost) < 0.10 and p_beta_zero > 0.05,
        'annualised_alpha': round(alpha_annual * 100, 2),  # in percent
        'information_ratio': round(information_ratio, 3),
        'sharpe_ratio': round(sharpe, 3),
        'r_squared': round(r_squared, 4),
        'tracking_error_annual': round(tracking_error * 100, 2),
        'rolling_beta_mean': round(np.mean(rolling_betas), 4) if rolling_betas else None,
        'rolling_beta_std': round(np.std(rolling_betas), 4) if rolling_betas else None,
        'max_beta_drift': round(max(abs(b) for b in rolling_betas), 4) if rolling_betas else None,
        'n_obs': n,
    }


def diagnose_neutrality_failure(results: dict) -> list[str]:
    """
    Interpret ex-post beta analysis results and suggest fixes.
    """
    issues = []

    if abs(results['ex_post_beta']) > 0.15:
        issues.append(
            f"High ex-post beta ({results['ex_post_beta']:.3f}). "
            "Likely cause: beta estimation lag or position sizing errors. "
            "Consider shorter beta estimation window (21–42 days) and more frequent rebalancing."
        )

    if results['rolling_beta_std'] and results['rolling_beta_std'] > 0.20:
        issues.append(
            f"High beta instability (std={results['rolling_beta_std']:.3f}). "
            "Individual stock betas are unstable. Switch to sector-level hedging or add futures overlay."
        )

    if results['r_squared'] > 0.30:
        issues.append(
            f"High market R² ({results['r_squared']:.3f}) — portfolio moves with market. "
            "Factor exposures (sector, size) may be causing residual correlation. "
            "Add explicit factor neutrality constraints to optimiser."
        )

    if results['annualised_alpha'] < 0:
        issues.append(
            "Negative alpha after market neutralisation. "
            "Alpha signals may be decaying faster than rebalance frequency. "
            "Consider increasing signal refresh rate or reducing holding period."
        )

    if not issues:
        issues.append("Portfolio appears well-neutralised. Continue monitoring for regime changes.")

    return issues

Purple Flea API Integration for Simultaneous Long/Short

Market-neutral strategies require sending long and short orders simultaneously to avoid market impact between the two legs. The Purple Flea trading API supports atomic pair execution, ensuring both legs fill at the same instant.

Authentication and Setup

Python — Purple Flea API Setup
import asyncio
import httpx
from typing import Literal

PF_API_BASE = "https://api.purpleflea.com/v1"
PF_API_KEY  = "pf_live_your_key_here"   # Never use sk_live_ prefix

class PurpleFleatradingClient:
    def __init__(self, api_key: str = PF_API_KEY):
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "X-Agent-Strategy": "market-neutral"
        }
        self.base = PF_API_BASE
        self.client = httpx.AsyncClient(headers=self.headers, timeout=15.0)

    async def place_order(
        self,
        ticker: str,
        side: Literal['buy', 'sell'],
        quantity: float,
        order_type: Literal['market', 'limit'] = 'market',
        limit_price: float = None,
        metadata: dict = None
    ) -> dict:
        payload = {
            "ticker": ticker,
            "side": side,
            "quantity": quantity,
            "order_type": order_type,
            "metadata": metadata or {}
        }
        if limit_price:
            payload["limit_price"] = limit_price

        r = await self.client.post(f"{self.base}/orders", json=payload)
        r.raise_for_status()
        return r.json()

    async def place_pair_orders(
        self,
        long_ticker: str,
        long_qty: float,
        short_ticker: str,
        short_qty: float,
        strategy_tag: str = "market_neutral"
    ) -> dict:
        """
        Submit long and short orders as an atomic pair.
        Purple Flea guarantees same-timestamp execution for pair orders.
        """
        payload = {
            "orders": [
                {
                    "ticker": long_ticker,
                    "side": "buy",
                    "quantity": long_qty,
                    "order_type": "market",
                    "pair_leg": "long"
                },
                {
                    "ticker": short_ticker,
                    "side": "sell",
                    "quantity": short_qty,
                    "order_type": "market",
                    "pair_leg": "short"
                }
            ],
            "execution_mode": "atomic_pair",
            "strategy": strategy_tag
        }
        r = await self.client.post(f"{self.base}/orders/pair", json=payload)
        r.raise_for_status()
        return r.json()

    async def close_pair(self, pair_id: str) -> dict:
        """Close all positions associated with a pair ID."""
        r = await self.client.post(f"{self.base}/orders/pair/{pair_id}/close")
        r.raise_for_status()
        return r.json()

    async def get_portfolio_beta(self) -> float:
        """Fetch current portfolio beta from Purple Flea risk engine."""
        r = await self.client.get(f"{self.base}/portfolio/risk")
        r.raise_for_status()
        data = r.json()
        return data.get("portfolio_beta", 0.0)

    async def get_positions(self) -> list[dict]:
        r = await self.client.get(f"{self.base}/portfolio/positions")
        r.raise_for_status()
        return r.json().get("positions", [])

    async def close(self):
        await self.client.aclose()

Full Agent Execution Loop

Python — Production Agent Loop
async def run_market_neutral_agent():
    """
    Full production loop: scan signals, construct portfolio,
    execute via Purple Flea, monitor beta drift, rebalance.
    """
    agent  = MarketNeutralAgent(api_key=PF_API_KEY)
    client = PurpleFleatradingClient(api_key=PF_API_KEY)

    print("[Agent] Starting market-neutral agent...")

    # Example universe (replace with your signal source)
    long_signals  = [
        {'ticker': 'AAPL', 'alpha_score': 0.82},
        {'ticker': 'MSFT', 'alpha_score': 0.78},
        {'ticker': 'NVDA', 'alpha_score': 0.71},
        {'ticker': 'GOOGL', 'alpha_score': 0.68},
        {'ticker': 'META', 'alpha_score': 0.65},
    ]
    short_signals = [
        {'ticker': 'IBM',   'alpha_score': -0.72},
        {'ticker': 'INTC',  'alpha_score': -0.68},
        {'ticker': 'HPQ',   'alpha_score': -0.61},
        {'ticker': 'DELL',  'alpha_score': -0.58},
        {'ticker': 'CSCO',  'alpha_score': -0.54},
    ]

    total_capital = 100_000  # $100K notional

    # Seed price history (in production, pull from market data feed)
    # await agent.update_prices({...})

    # Construct initial portfolio
    orders = agent.construct_portfolio(long_signals, short_signals, total_capital, n_pairs=5)
    print(f"[Agent] Constructed {len(orders)} orders")

    # Execute in pairs
    for i in range(0, len(orders), 2):
        if i + 1 >= len(orders):
            break
        long_ord  = orders[i]
        short_ord = orders[i+1]

        try:
            result = await client.place_pair_orders(
                long_ticker  = long_ord['ticker'],
                long_qty     = long_ord['quantity'],
                short_ticker = short_ord['ticker'],
                short_qty    = short_ord['quantity'],
                strategy_tag = "market_neutral_v1"
            )
            print(f"[Agent] Pair executed: {long_ord['ticker']}/{short_ord['ticker']} → {result['pair_id']}")
        except Exception as e:
            print(f"[Agent] Order error: {e}")

    # Monitoring loop: check beta every 5 minutes
    while True:
        await asyncio.sleep(300)

        needs_rebalance, current_beta = agent.needs_rebalance()
        summary = agent.portfolio_summary()
        print(f"[Agent] Beta: {current_beta:.4f} | PnL: ${summary['total_pnl']:,.2f}")

        if needs_rebalance:
            print(f"[Agent] Beta drift detected ({current_beta:.4f}) — computing hedge...")
            hedge_order = await agent.get_beta_hedge(
                target_beta_delta=current_beta,
                portfolio_value=total_capital
            )
            if hedge_order['quantity'] > 0:
                result = await client.place_order(
                    ticker     = hedge_order['instrument'],
                    side       = hedge_order['side'],
                    quantity   = hedge_order['quantity'],
                    order_type = 'market',
                    metadata   = {'purpose': 'beta_hedge', 'target_beta': 0.0}
                )
                print(f"[Agent] Hedge placed: {hedge_order['quantity']} {hedge_order['instrument']} {hedge_order['side']}")
            else:
                print(f"[Agent] Beta drift too small for minimum futures lot — monitoring.")

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

Key Purple Flea API Endpoints

Risk Management for Market-Neutral Agents

Market-neutral does not mean risk-free. The key risks that remain after eliminating beta are:

Conclusion

Market-neutral strategies represent the most sophisticated form of alpha harvesting available to AI trading agents. By eliminating beta exposure through precise beta calculation, factor-neutral portfolio construction, and futures-based hedging overlays, agents can generate uncorrelated returns regardless of market direction.

The key components covered in this guide:

• Zero-beta portfolio theory: understanding what neutrality actually means mathematically and operationally.
• Factor-neutral construction: neutralising not just market beta but value, momentum, size, and quality exposures via QP optimisation.
• Statistical arbitrage: cointegration-based mean-reverting spread trading with full Z-score signal generation.
• Volatility-neutral options: delta-hedged positions that extract pure volatility premium.
• Beta hedging with futures: using ES and MES futures as the fastest and cheapest neutralisation instrument.
• Ex-ante vs ex-post analysis: measuring and diagnosing neutrality failure in realised returns.
• Purple Flea API integration: atomic pair execution that eliminates implementation shortfall between long and short legs.

Agents that master market-neutral execution can operate in any market regime — bear, bull, or sideways — generating consistent alpha while remaining immune to systematic risk. The Purple Flea trading API provides the infrastructure to execute these strategies at production scale.

Related Reading

• Options Greeks for AI Agents: Delta, Gamma, Vega, Theta
• Statistical Arbitrage: Complete Agent Implementation
• Cross-Exchange Arbitrage for AI Agents
• Agent Portfolio Theory: Constructing Alpha-Optimal Books
• Risk Management Frameworks for Autonomous Agents
• High-Frequency Trading for AI Agents
• Perpetual Funding Rates: Agent Strategies