Advanced Grid Bot Strategies for AI Agents

1. Arithmetic vs Geometric Grids: The Fundamental Choice

The most important architectural decision in any grid bot is whether to use arithmetic (linear) grids or geometric (logarithmic) grids. This single choice determines profit distribution, capital efficiency, and performance characteristics across different price ranges.

Grid Spacing Mathematics

Understanding the mathematical basis for each grid type enables agents to calculate precise profit expectations before deployment.

import numpy as np
from typing import List, Tuple

def arithmetic_grid(lower: float, upper: float, n_levels: int) -> List[float]:
    """Generate arithmetic grid levels with equal dollar spacing."""
    spacing = (upper - lower) / (n_levels - 1)
    return [round(lower + i * spacing, 6) for i in range(n_levels)]

def geometric_grid(lower: float, upper: float, n_levels: int) -> List[float]:
    """Generate geometric grid levels with equal percentage spacing."""
    ratio = (upper / lower) ** (1 / (n_levels - 1))
    return [round(lower * ratio**i, 6) for i in range(n_levels)]

def grid_spacing_pct(levels: List[float]) -> float:
    """Average percentage spacing between adjacent levels."""
    spacings = [(levels[i+1 ] - levels[i]) / levels[i] * 100 for i in range(len(levels)-1)]
    return np.mean(spacings)

def profit_per_grid(
    buy_price: float,
    sell_price: float,
    quantity: float,
    fee_rate: float = 0.001   # 0.1% maker fee
) -> float:
    """Profit from one completed grid cycle (buy at lower, sell at upper)."""
    gross = (sell_price - buy_price) * quantity
    fees = (buy_price + sell_price) * quantity * fee_rate
    return gross - fees

# Compare grid types on BTC-like asset
arith = arithmetic_grid(40000, 50000, 20)
geo   = geometric_grid(40000, 50000, 20)

print(ff"Arithmetic spacing: {grid_spacing_pct(arith):.2f}% avg")
print(ff"Geometric spacing:  {grid_spacing_pct(geo):.4f}% (constant)")

2. Capital Efficiency and Grid Sizing Math

Capital efficiency determines how effectively deployed capital is utilized. A poorly designed grid can leave large portions of capital idle while a small fraction does all the work.

Capital Allocation Across Grid Levels

For a grid bot that always has buy orders below current price and sell orders above, capital allocation depends on the current price position within the range and the grid width.

def calculate_grid_capital(
    lower: float,
    upper: float,
    n_levels: int,
    total_capital: float,
    grid_type: str = 'geometric'
) -> dict:
    """
    Calculate optimal per-grid allocation and expected returns.
    Assumes equal capital per buy level (quote currency reserved).
    """
    if grid_type == 'geometric':
        levels = geometric_grid(lower, upper, n_levels)
    else:
        levels = arithmetic_grid(lower, upper, n_levels)

    n_buy_levels = n_levels // 2    # buy orders below mid, sell above
    capital_per_level = total_capital / n_buy_levels
    qty_per_level = [capital_per_level / price for price in levels[:n_buy_levels]]

    # Profit per grid cycle
    profits = []
    for i in range(n_levels - 1):
        p = profit_per_grid(levels[i], levels[i+1], capital_per_level / levels[i])
        profits.append(p)

    spacing = grid_spacing_pct(levels)
    annual_cycles_estimate = (365 * 24 * 2) / (spacing / 100 * 10000)  # rough estimate

    return {
        'levels': levels,
        'spacing_pct': spacing,
        'capital_per_level': capital_per_level,
        'avg_profit_per_cycle': np.mean(profits),
        'total_profit_if_all_cycle': sum(profits),
        'estimated_annual_return_pct': (sum(profits) / total_capital) * 100
    }

# Example: $10,000 capital, 20 levels, geometric grid
result = calculate_grid_capital(40000, 50000, 20, 10000, 'geometric')
print(ff"Spacing: {result['spacing_pct']:.3f}% per grid")
print(ff"Profit per cycle: ${result['avg_profit_per_cycle']:.4f}")

3. ATR-Based Dynamic Range Selection

The Average True Range (ATR) is the gold standard for measuring market volatility and selecting grid ranges. A grid too narrow will be breached constantly, requiring costly resets. A grid too wide will be capital-inefficient with orders too rarely filled.

ATR Calculation and Application

import pandas as pd
import numpy as np

def calculate_atr(highs: pd.Series, lows: pd.Series, closes: pd.Series, period: int = 14) -> pd.Series:
    """Average True Range over the specified period."""
    tr1 = highs - lows
    tr2 = (highs - closes.shift(1)).abs()
    tr3 = (lows - closes.shift(1)).abs()
    true_range = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    return true_range.rolling(period).mean()

def select_grid_range(
    current_price: float,
    atr: float,
    atr_multiplier_lower: float = 2.0,
    atr_multiplier_upper: float = 2.0,
    snap_to_round: bool = True
) -> Tuple[float, float]:
    """
    Select grid upper and lower bounds using ATR.
    2x ATR below and above current price = ~95% of typical daily moves captured.
    """
    lower = current_price - (atr * atr_multiplier_lower)
    upper = current_price + (atr * atr_multiplier_upper)

    if snap_to_round:
        # Round to nearest significant price level for cleaner order book placement
        magnitude = 10 ** (int(np.log10(current_price)) - 1)
        lower = round(lower / magnitude) * magnitude
        upper = round(upper / magnitude) * magnitude

    return lower, upper

def range_validity_check(lower: float, upper: float, current_price: float, min_range_pct: float = 5.0) -> bool:
    """Validate that the selected range is wide enough to be meaningful."""
    range_pct = (upper - lower) / current_price * 100
    price_in_range = lower < current_price < upper
    return range_pct >= min_range_pct and price_in_range

Multi-Timeframe ATR Blending

A more sophisticated approach blends ATR values from multiple timeframes. Short-term ATR captures intraday swings while long-term ATR captures macro range tendencies.

def blended_atr_range(price: float, df_1h: pd.DataFrame, df_4h: pd.DataFrame, df_1d: pd.DataFrame) -> Tuple[float, float]:
    """Blend ATR across 1h, 4h, 1d timeframes for robust range selection."""
    atr_1h = calculate_atr(df_1h['high'], df_1h['low'], df_1h['close']).iloc[-1]
    atr_4h = calculate_atr(df_4h['high'], df_4h['low'], df_4h['close']).iloc[-1]
    atr_1d = calculate_atr(df_1d['high'], df_1d['low'], df_1d['close']).iloc[-1]

    # Weighted average: higher weight to slower timeframes for grid range
    blended_atr = (0.2 * atr_1h + 0.3 * atr_4h + 0.5 * atr_1d)
    return select_grid_range(price, blended_atr, atr_multiplier_lower=2.5, atr_multiplier_upper=2.5)

4. Market Regime Detection: Range-Bound vs Trending

Grid bots perform excellently in range-bound markets and catastrophically in strong trends. An agent must detect market regime before deploying a grid and pause or close the grid during trending conditions.

Regime Detection Indicators

from enum import Enum

class MarketRegime(Enum):
    RANGING   = "ranging"
    TRENDING  = "trending"
    VOLATILE  = "volatile"
    UNCERTAIN = "uncertain"

class RegimeDetector:
    def __init__(self, df: pd.DataFrame):
        self.df = df
        self.close = df['close']

    def adx(self, period: int = 14) -> float:
        """Average Directional Index. >25 = trending, <20 = ranging."""
        high = self.df['high']
        low  = self.df['low']
        up_move   = high - high.shift(1)
        down_move = low.shift(1) - low
        plus_dm  = np.where((up_move > down_move) & (up_move > 0), up_move, 0)
        minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0)
        tr = calculate_atr(high, low, self.close, period)
        plus_di  = 100 * pd.Series(plus_dm).rolling(period).mean() / tr
        minus_di = 100 * pd.Series(minus_dm).rolling(period).mean() / tr
        dx = (abs(plus_di - minus_di) / (plus_di + minus_di + 1e-10)) * 100
        return dx.rolling(period).mean().iloc[-1]

    def bollinger_width(self, period: int = 20) -> float:
        """Normalized Bollinger Band width. Low = squeeze = potential range."""
        sma = self.close.rolling(period).mean()
        std = self.close.rolling(period).std()
        upper = sma + 2 * std
        lower = sma - 2 * std
        width = (upper - lower) / sma
        return width.iloc[-1]

    def detect(self) -> MarketRegime:
        adx_val = self.adx()
        bw = self.bollinger_width()

        if adx_val < 20 and bw < 0.05:
            return MarketRegime.RANGING
        elif adx_val > 30:
            return MarketRegime.TRENDING
        elif bw > 0.15:
            return MarketRegime.VOLATILE
        return MarketRegime.UNCERTAIN

5. Stop-Loss Mechanisms and Grid Reset Logic

Grid bots without stop-losses become bagholders. When price breaks out of the range, the bot accumulates a full position at the breached edge with unrealized losses continuing to grow. Stop-loss logic must be embedded in the agent's core decision loop.

Three-Layer Stop Architecture

• Price stop: If price closes below the lower grid bound (or above upper), immediately cancel all orders and evaluate whether to reset.
• Drawdown stop: If total unrealized + realized PnL drops below -X% of deployed capital, shut down and alert.
• Time stop: If the grid has been running for N days without a full cycle on any level, reassess range suitability.

class GridStopLossManager:
    def __init__(self, lower: float, upper: float, total_capital: float):
        self.lower = lower
        self.upper = upper
        self.total_capital = total_capital
        self.max_drawdown_pct = 0.15     # 15% max drawdown
        self.time_stop_days = 14         # no cycles in 14 days = stale grid
        self.last_cycle_time = None

    def check_price_stop(self, current_price: float) -> bool:
        return current_price < self.lower or current_price > self.upper

    def check_drawdown_stop(self, realized_pnl: float, unrealized_pnl: float) -> bool:
        total_pnl = realized_pnl + unrealized_pnl
        drawdown_pct = -total_pnl / self.total_capital
        return drawdown_pct > self.max_drawdown_pct

    def check_time_stop(self) -> bool:
        from datetime import datetime
        if self.last_cycle_time is None:
            return False
        days_since = (datetime.now() - self.last_cycle_time).days
        return days_since > self.time_stop_days

    def should_stop(self, price: float, realized: float, unrealized: float) -> dict:
        reasons = []
        if self.check_price_stop(price):
            reasons.append("price_out_of_range")
        if self.check_drawdown_stop(realized, unrealized):
            reasons.append("max_drawdown_exceeded")
        if self.check_time_stop():
            reasons.append("time_stop_no_activity")
        return {'stop': bool(reasons), 'reasons': reasons}

6. Compound Grids and Reinvestment Logic

A basic grid bot keeps profits static. A compound grid reinvests realized profits back into the grid, gradually increasing position size and accelerating returns. This is the grid equivalent of compound interest.

class CompoundGridEngine:
    """
    Automatically reinvests grid profits into additional grid levels
    or increases quantity per level as profits accumulate.
    """
    def __init__(self, initial_capital: float, compound_threshold_pct: float = 2.0):
        self.capital = initial_capital
        self.realized_pnl = 0.0
        self.compound_threshold = initial_capital * compound_threshold_pct / 100
        self.reinvested_total = 0.0
        self.cycle_count = 0

    def record_cycle_profit(self, profit: float):
        self.realized_pnl += profit
        self.cycle_count += 1

    def should_compound(self) -> bool:
        return self.realized_pnl >= self.compound_threshold

    def compound(self, grid_levels: list, current_price: float) -> dict:
        """Reinvest accumulated profits by increasing per-level allocation."""
        if not self.should_compound():
            return {'compounded': False}

        amount_to_add = self.realized_pnl
        self.capital += amount_to_add
        self.reinvested_total += amount_to_add
        self.realized_pnl = 0.0

        n_buy_levels = len([l for l in grid_levels if l < current_price])
        new_capital_per_level = self.capital / n_buy_levels

        return {
            'compounded': True,
            'new_capital': self.capital,
            'reinvested': amount_to_add,
            'new_capital_per_level': new_capital_per_level,
            'total_reinvested': self.reinvested_total
        }

7. Full AdvancedGridBot Implementation

The complete agent integrates all the components above: regime detection, ATR-based range selection, geometric grid generation, stop-loss management, and compound reinvestment.

import asyncio
import logging
import httpx
from datetime import datetime

class AdvancedGridBot:
    """
    Fully autonomous grid trading agent with:
    - ATR-based dynamic range selection
    - Market regime detection (only runs in ranging markets)
    - Geometric grid spacing for crypto assets
    - Compound reinvestment on profit milestones
    - Multi-layer stop-loss protection
    """

    TRADING_API = "https://trading.purpleflea.com"
    WALLET_API  = "https://wallet.purpleflea.com"

    def __init__(self, api_key: str, symbol: str, capital: float, n_grids: int = 30):
        self.api_key = api_key
        self.symbol = symbol
        self.capital = capital
        self.n_grids = n_grids
        self.active_orders = {}
        self.regime = MarketRegime.UNCERTAIN
        self.grid_levels = []
        self.stop_manager = None
        self.compound_engine = CompoundGridEngine(capital)
        self.logger = logging.getLogger(ff'GridBot.{symbol}')
        self.client = httpx.AsyncClient(
            headers={'Authorization': ff'Bearer {api_key}'}, timeout=30
        )

    async def get_market_data(self) -> pd.DataFrame:
        """Fetch OHLCV candles for regime detection and ATR calculation."""
        resp = await self.client.get(
            ff'{self.TRADING_API}/candles/{self.symbol}',
            params={'interval': '4h', 'limit': 100}
        )
        data = resp.json()
        return pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

    async def initialize_grid(self):
        """Detect regime, select range, generate levels, and place all orders."""
        df = await self.get_market_data()
        detector = RegimeDetector(df)
        self.regime = detector.detect()

        if self.regime == MarketRegime.TRENDING:
            self.logger.warning("Trending regime detected — grid not deployed.")
            return False

        current_price = df['close'].iloc[-1]
        atr = calculate_atr(df['high'], df['low'], df['close']).iloc[-1]
        lower, upper = select_grid_range(current_price, atr, 2.5, 2.5)

        if not range_validity_check(lower, upper, current_price):
            self.logger.error("Range validation failed — skipping deployment.")
            return False

        self.grid_levels = geometric_grid(lower, upper, self.n_grids)
        self.stop_manager = GridStopLossManager(lower, upper, self.capital)

        # Place buy orders below current price, sell orders above
        capital_per_level = self.capital / (self.n_grids // 2)
        await self.place_grid_orders(self.grid_levels, current_price, capital_per_level)

        self.logger.info(ff"Grid deployed: [{lower:.2f}, {upper:.2f}] | {self.n_grids} levels | ATR: {atr:.2f}")
        return True

    async def place_grid_orders(self, levels: list, current_price: float, capital_per_level: float):
        for i, level in enumerate(levels):
            qty = capital_per_level / level
            side = 'buy' if level < current_price else 'sell'
            order = {
                'symbol': self.symbol, 'side': side,
                'type': 'limit', 'price': level,
                'quantity': round(qty, 6), 'grid_level': i
            }
            resp = await self.client.post(ff'{self.TRADING_API}/orders', json=order)
            self.active_orders[i] = resp.json()

    async def on_fill(self, fill_event: dict):
        """Handle order fill: place counter-order and check compound."""
        level_idx = fill_event['grid_level']
        side = fill_event['side']
        profit = fill_event.get('realized_pnl', 0)

        if profit > 0:
            self.compound_engine.record_cycle_profit(profit)
            result = self.compound_engine.compound(self.grid_levels, fill_event['price'])
            if result['compounded']:
                self.logger.info(ff"Compounded! New capital: ${result['new_capital']:.2f}")

        # Place counter order on the adjacent level
        counter_side = 'sell' if side == 'buy' else 'buy'
        counter_level_idx = level_idx + 1 if side == 'buy' else level_idx - 1
        if 0 <= counter_level_idx < len(self.grid_levels):
            counter_price = self.grid_levels[counter_level_idx]
            qty = fill_event['quantity']
            order = {'symbol': self.symbol, 'side': counter_side, 'type': 'limit',
                     'price': counter_price, 'quantity': qty, 'grid_level': counter_level_idx}
            resp = await self.client.post(ff'{self.TRADING_API}/orders', json=order)
            self.active_orders[counter_level_idx] = resp.json()

    async def run(self, poll_seconds: int = 60):
        started = await self.initialize_grid()
        if not started:
            self.logger.info("Grid not started — waiting for regime change.")
            return

        while True:
            try:
                status = await self.client.get(ff'{self.TRADING_API}/grid-status/{self.symbol}')
                data = status.json()
                stop = self.stop_manager.should_stop(
                    data['current_price'], data['realized_pnl'], data['unrealized_pnl']
                )
                if stop['stop']:
                    self.logger.warning(ff"Grid stop triggered: {stop['reasons']}")
                    await self.shutdown()
                    break
            except Exception as e:
                self.logger.error(ff'Poll error: {e}')
            await asyncio.sleep(poll_seconds)

    async def shutdown(self):
        for order_id in self.active_orders.values():
            await self.client.delete(ff'{self.TRADING_API}/orders/{order_id}')
        self.active_orders.clear()
        self.logger.info("Grid shut down. All orders cancelled.")

8. Performance Metrics and Optimization

Measuring grid bot performance requires metrics specific to range-based strategies. Standard metrics like Sharpe ratio are insufficient on their own.

9. Multi-Symbol Grid Portfolio Architecture

Running a single grid bot on one symbol is a good start. Running a coordinated portfolio of grid bots across multiple uncorrelated symbols with shared capital management is significantly more capital-efficient.

Portfolio Grid Strategy

A portfolio approach allocates a fixed capital pool across multiple symbols, dynamically rebalancing as different grids complete or stall. Symbols with high cycle rates receive more capital; stale grids get reduced allocation.

class GridPortfolioManager:
    """
    Manages a pool of AdvancedGridBot instances across multiple symbols.
    Dynamically reallocates capital based on cycle performance.
    """
    def __init__(self, api_key: str, total_capital: float, symbols: List[str]):
        self.api_key = api_key
        self.total_capital = total_capital
        self.symbols = symbols
        self.bots: Dict[str, AdvancedGridBot] = {}
        self.allocation: Dict[str, float] = {}
        self.cycle_counts: Dict[str, int] = {s: 0 for s in symbols}

    def initial_allocation(self) -> Dict[str, float]:
        """Equal weight initially; reserves 20% as buffer."""
        deployable = self.total_capital * 0.80
        per_symbol = deployable / len(self.symbols)
        return {s: per_symbol for s in self.symbols}

    def rebalance(self) -> Dict[str, float]:
        """Reallocate capital proportional to cycle counts (performance-weighted)."""
        total_cycles = sum(self.cycle_counts.values()) + len(self.symbols)  # +1 each to avoid zero
        deployable = self.total_capital * 0.80
        new_alloc = {}
        for s in self.symbols:
            weight = (self.cycle_counts[s] + 1) / total_cycles
            new_alloc[s] = deployable * weight
        return new_alloc

    async def run_portfolio(self):
        """Start all bots concurrently, rebalance weekly."""
        alloc = self.initial_allocation()
        tasks = []
        for symbol in self.symbols:
            bot = AdvancedGridBot(self.api_key, symbol, alloc[symbol])
            self.bots[symbol] = bot
            tasks.append(asyncio.create_task(bot.run()))

        # Rebalance every 7 days
        async def rebalancer():
            while True:
                await asyncio.sleep(7 * 24 * 3600)
                new_alloc = self.rebalance()
                for s, capital in new_alloc.items():
                    self.bots[s].capital = capital

        tasks.append(asyncio.create_task(rebalancer()))
        await asyncio.gather(*tasks)

10. Grid Bot Tuning Parameters Reference

The following table summarizes the key tuning parameters for AdvancedGridBot and their recommended starting ranges for different market conditions and asset classes.

11. Funding Rate Integration for Perpetual Grid Bots

On perpetual futures markets like those available at trading.purpleflea.com, funding rates add another dimension to grid bot profitability. When funding rates are significantly positive (longs pay shorts), a grid bot that holds short positions between cycles earns funding income. When negative, it pays.

Funding Rate Aware Grid Logic

An advanced grid bot on perpetual markets should factor the 8-hour funding rate into its profitability model:

async def get_funding_rate(self, symbol: str) -> float:
    """Fetch current funding rate for a perpetual market."""
    resp = await self.client.get(
        ff'{self.TRADING_API}/funding-rate/{symbol}'
    )
    data = resp.json()
    # Annualize: 8h rate * 3 * 365
    annualized = data['rate_8h'] * 3 * 365
    return annualized

def funding_adjusted_profit_per_cycle(
    grid_profit: float,
    avg_position_size: float,
    avg_hold_hours: float,
    funding_rate_8h: float
) -> float:
    """Adjust grid cycle profit by expected funding payments/receipts."""
    funding_per_8h = avg_position_size * funding_rate_8h
    periods = avg_hold_hours / 8
    funding_income = funding_per_8h * periods   # positive if short and rate positive
    return grid_profit + funding_income

def should_deploy_given_funding(
    annualized_funding: float,
    grid_annual_return: float,
    side_bias: str  # 'long' or 'short'
) -> bool:
    """
    Only deploy grid if net return (grid + funding) is positive.
    In high positive funding environments, short-biased grids get a tailwind.
    In high negative funding, long-biased grids benefit.
    """
    if side_bias == 'short':
        net = grid_annual_return + annualized_funding   # shorts receive positive funding
    else:
        net = grid_annual_return - annualized_funding   # longs pay positive funding
    return net > 0.05   # require at least 5% net annual return