Mean Reversion Trading for AI Agents: Statistical Edge in Crypto Markets

Crypto markets are noisier than any other asset class on earth. Prices swing 20% in a weekend. Retail traders panic-buy tops and panic-sell bottoms. This irrationality is not a bug for AI agents — it is a feature. Where humans see chaos, statistical agents see reversion to the mean. This guide breaks down the indicators, math, and code for building a profitable mean reversion system on Purple Flea Trading.

1. The Statistical Basis of Mean Reversion

Mean reversion is rooted in the concept of stationarity: some time series have a tendency to return to a long-run average after deviating from it. Prices themselves are not stationary (they trend). But price spreads, ratios, and oscillators derived from prices often are.

The formal test is the Augmented Dickey-Fuller (ADF) test for unit roots. If a price series rejects the null hypothesis of a unit root at 95% confidence, it is stationary — mean-reverting. Most raw crypto prices fail this test. But the spread between BTC and ETH, or the deviation of price from its 20-day moving average, often passes.

A half-life of 6 hours means that, on average, a deviation from the mean is expected to halve within 6 hours. This directly informs position holding time and stop-loss distances.

2. Bollinger Bands: Price Envelope Signals

Bollinger Bands are the most widely used mean reversion indicator. They plot a moving average with upper and lower bands at N standard deviations — creating a dynamic envelope that expands during volatility and contracts during calm periods.

Signal Logic

• Price touches lower band: Potential long entry — price is statistically stretched below average
• Price returns to middle band (SMA): Profit target for long positions
• Price touches upper band: Potential short entry — price is statistically stretched above average
• Band squeeze (low width): Volatility compression, often precedes a breakout — temporarily suspend mean reversion signals

import numpy as np
import pandas as pd

def bollinger_bands(prices: pd.Series, n: int = 20, k: float = 2.0) -> pd.DataFrame:
    """
    Compute Bollinger Bands for a price series.
    Returns DataFrame with: sma, upper, lower, pct_b, bandwidth
    """
    sma = prices.rolling(n).mean()
    std = prices.rolling(n).std()
    upper = sma + k * std
    lower = sma - k * std

    # %B: where price sits within the bands (0=lower, 1=upper, 0.5=middle)
    pct_b = (prices - lower) / (upper - lower)

    # Bandwidth: normalized band width (indicator of volatility regime)
    bandwidth = (upper - lower) / sma

    return pd.DataFrame({
        "sma": sma, "upper": upper, "lower": lower,
        "pct_b": pct_b, "bandwidth": bandwidth
    })

def bollinger_signal(bb: pd.DataFrame, price: float, bandwidth_threshold: float = 0.02) -> str:
    """
    Generate trading signal from Bollinger Band data.
    bandwidth_threshold: minimum band width (avoids trading during squeeze)
    """
    latest = bb.iloc[-1]

    if latest.bandwidth < bandwidth_threshold:
        return "squeeze_no_trade"

    if latest.pct_b < 0.05:   # Price near/below lower band
        return "long"
    elif latest.pct_b > 0.95:  # Price near/above upper band
        return "short"
    elif 0.45 < latest.pct_b < 0.55:
        return "mean_exit"  # Close position: price returned to mean
    else:
        return "hold"

3. RSI: Momentum Exhaustion Signals

The Relative Strength Index (RSI) measures the speed and magnitude of recent price changes on a 0-100 scale. Developed by J. Welles Wilder, it excels at identifying momentum exhaustion — points where a move has gone too far, too fast.

RSI Signal Thresholds

def compute_rsi(prices: pd.Series, n: int = 14) -> pd.Series:
    """Compute RSI for a price series."""
    delta = prices.diff()
    gain = delta.clip(lower=0)
    loss = (-delta).clip(lower=0)

    avg_gain = gain.ewm(com=n-1, adjust=False).mean()
    avg_loss = loss.ewm(com=n-1, adjust=False).mean()

    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

def rsi_signal(rsi: float, oversold: float = 30, overbought: float = 70) -> str:
    if rsi < oversold:
        strength = "strong" if rsi < 20 else "moderate"
        return f"long_{strength}"
    elif rsi > overbought:
        strength = "strong" if rsi > 80 else "moderate"
        return f"short_{strength}"
    else:
        return "neutral"

4. Z-Score Entry and Exit Logic

The z-score is the most statistically rigorous mean reversion entry signal. It measures how many standard deviations the current price is from its historical mean — giving you a direct probabilistic interpretation of the extremity of the current deviation.

Under a normal distribution: |Z| > 1.64 occurs only 10% of the time, |Z| > 1.96 only 5%, |Z| > 2.58 only 1%. These levels map naturally to entry thresholds of increasing confidence.

def compute_zscore(prices: pd.Series, lookback: int = 20) -> pd.Series:
    """Compute rolling z-score of price."""
    mu = prices.rolling(lookback).mean()
    sigma = prices.rolling(lookback).std()
    return (prices - mu) / sigma

def zscore_signal(z: float, entry_threshold: float = 2.0, exit_threshold: float = 0.5) -> str:
    """
    Generate trading signal from z-score.

    entry_threshold: z-score level to enter (|z| > entry_threshold)
    exit_threshold:  z-score level to exit  (|z| < exit_threshold)
    """
    if z < -entry_threshold:
        return "long_entry"
    elif z > entry_threshold:
        return "short_entry"
    elif abs(z) < exit_threshold:
        return "exit"   # Price returned to mean — close position
    else:
        return "hold"

# Z-score signal table
examples = [
    ("Deep oversold",  -3.2),  # Very high confidence long
    ("Oversold",       -2.1),  # Standard long entry
    ("Slightly low",   -1.3),  # Hold or partial position
    ("Near mean",       0.2),  # Exit: price returned to average
    ("Overbought",      2.3),  # Short entry
    ("Extreme high",    3.8),  # Strong short signal
]

for label, z in examples:
    sig = zscore_signal(z)
    print(f"z={z:+.1f} ({label}): {sig}")

5. Combining Signals: Bollinger + RSI + Z-Score

Individual indicators generate false signals. The most robust mean reversion systems require confluence — multiple independent indicators agreeing before entering a trade. This dramatically increases win rate while reducing trade frequency.

Confluence Entry Rules

• Long entry: Bollinger %B < 0.1 AND RSI < 35 AND Z-score < -1.8
• Short entry: Bollinger %B > 0.9 AND RSI > 65 AND Z-score > 1.8
• Exit: Price crosses SMA (Bollinger middle band) OR Z-score crosses ±0.3

def combined_signal(prices: pd.Series) -> dict:
    """
    Combine Bollinger, RSI, and Z-score into a single confluence signal.
    Returns signal dict with action, confidence, and individual indicator values.
    """
    bb = bollinger_bands(prices)
    rsi = compute_rsi(prices)
    zscore = compute_zscore(prices)

    latest_bb = bb.iloc[-1]
    latest_rsi = rsi.iloc[-1]
    latest_z = zscore.iloc[-1]

    # Count bullish conditions
    bullish = sum([
        latest_bb.pct_b < 0.1,
        latest_rsi < 35,
        latest_z < -1.8,
    ])

    # Count bearish conditions
    bearish = sum([
        latest_bb.pct_b > 0.9,
        latest_rsi > 65,
        latest_z > 1.8,
    ])

    if bullish >= 2:
        action = "long"
        confidence = "high" if bullish == 3 else "medium"
    elif bearish >= 2:
        action = "short"
        confidence = "high" if bearish == 3 else "medium"
    elif latest_bb.bandwidth < 0.02:
        action = "no_trade"
        confidence = "squeeze"
    else:
        action = "hold"
        confidence = "none"

    return {
        "action": action,
        "confidence": confidence,
        "pct_b": round(latest_bb.pct_b, 3),
        "rsi": round(latest_rsi, 1),
        "zscore": round(latest_z, 2),
        "bandwidth": round(latest_bb.bandwidth, 4),
    }

6. The MeanReversionAgent Class

Here is a complete, production-ready agent that integrates all three signals and executes trades via the Purple Flea Trading API:

import asyncio
import requests
import pandas as pd
from datetime import datetime

class MeanReversionAgent:
    def __init__(
        self,
        api_key: str,
        symbols: list,
        capital_per_symbol: float = 1000.0,
        bb_n: int = 20,
        bb_k: float = 2.0,
        rsi_n: int = 14,
        zscore_lookback: int = 20,
        max_hold_hours: int = 24
    ):
        self.api_key = api_key
        self.symbols = symbols
        self.capital = capital_per_symbol
        self.bb_n = bb_n
        self.bb_k = bb_k
        self.rsi_n = rsi_n
        self.zscore_lookback = zscore_lookback
        self.max_hold_hours = max_hold_hours
        self.positions = {}  # {symbol: {side, entry_price, entry_time, size}}
        self.trades_log = []
        self.base_url = "https://api.purpleflea.com/trading/v1"
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def fetch_ohlcv(self, symbol: str, interval: str = "1h", limit: int = 100) -> pd.DataFrame:
        """Fetch OHLCV candlestick data from Purple Flea Trading."""
        r = requests.get(
            f"{self.base_url}/candles",
            headers=self.headers,
            params={"symbol": symbol, "interval": interval, "limit": limit}
        )
        data = r.json()
        df = pd.DataFrame(data, columns=["ts", "open", "high", "low", "close", "volume"])
        df["close"] = df["close"].astype(float)
        return df

    def place_order(self, symbol: str, side: str, usdc_size: float) -> dict:
        """Place a market order on Purple Flea Trading."""
        r = requests.post(
            f"{self.base_url}/order",
            headers=self.headers,
            json={"symbol": symbol, "side": side, "type": "market", "quote_amount": usdc_size}
        )
        return r.json()

    def check_max_hold_exit(self, symbol: str):
        """Force-exit positions held beyond max_hold_hours."""
        pos = self.positions.get(symbol)
        if not pos: return
        hours_held = (datetime.utcnow() - pos["entry_time"]).total_seconds() / 3600
        if hours_held > self.max_hold_hours:
            close_side = "sell" if pos["side"] == "buy" else "buy"
            self.place_order(symbol, close_side, pos["size"])
            del self.positions[symbol]
            print(f"[{symbol}] Max hold time exceeded — force closed after {hours_held:.1f}h")

    async def run(self, interval_seconds: int = 3600):
        """Main agent loop. Evaluates signals every interval_seconds."""
        print(f"MeanReversionAgent started. Watching: {', '.join(self.symbols)}")

        while True:
            for symbol in self.symbols:
                try:
                    df = self.fetch_ohlcv(symbol)
                    signal = combined_signal(df["close"])
                    action = signal["action"]

                    self.check_max_hold_exit(symbol)

                    existing = self.positions.get(symbol)

                    if existing:
                        if action == "hold" or (action != existing["side"] and action in ["long", "short"]):
                            close_side = "sell" if existing["side"] == "long" else "buy"
                            self.place_order(symbol, close_side, existing["size"])
                            del self.positions[symbol]
                            print(f"[{symbol}] Closed {existing['side']} | signal={action} | z={signal['zscore']}")

                    elif action in ["long", "short"] and signal["confidence"] in ["medium", "high"]:
                        order_side = "buy" if action == "long" else "sell"
                        size_mult = 1.5 if signal["confidence"] == "high" else 1.0
                        size = self.capital * size_mult
                        order = self.place_order(symbol, order_side, size)
                        self.positions[symbol] = {
                            "side": action,
                            "entry_time": datetime.utcnow(),
                            "size": size,
                            "signal": signal,
                        }
                        print(f"[{symbol}] Opened {action} | confidence={signal['confidence']} | z={signal['zscore']} | rsi={signal['rsi']}")

                except Exception as e:
                    print(f"[{symbol}] Error: {e}")

            await asyncio.sleep(interval_seconds)

# Launch the agent
if __name__ == "__main__":
    agent = MeanReversionAgent(
        api_key="pf_live_your_key_here",
        symbols=["BTC/USDC", "ETH/USDC", "SOL/USDC"],
        capital_per_symbol=2000
    )
    asyncio.run(agent.run())

7. Regime Filter: When NOT to Mean-Revert

Mean reversion fails catastrophically during trending markets. When BTC is in a strong uptrend, "oversold" RSI readings just keep getting more oversold. A trending regime filter prevents you from fighting the tape:

def detect_regime(prices: pd.Series) -> str:
    """
    Detect market regime: trending vs. mean-reverting.
    Uses ADX-equivalent: ratio of directional to total movement.
    """
    short_ma = prices.rolling(10).mean()
    long_ma  = prices.rolling(50).mean()

    # Trend strength: distance between MAs relative to volatility
    trend_strength = abs(short_ma - long_ma) / prices.rolling(20).std()

    latest = trend_strength.iloc[-1]

    if latest > 1.5:
        return "trending"       # Strong trend: suppress mean reversion
    elif latest > 0.8:
        return "mixed"          # Ambiguous: reduce position sizes
    else:
        return "mean_reverting"  # Range-bound: full mean reversion signals

8. Performance Benchmarks and Expectations

Fewer, higher-quality trades driven by strong confluence signals consistently outperform high-frequency systems that trade every minor oscillation. The regime filter reduces trade count by ~40% while typically improving Sharpe by 50-80%.