Liquidation Protection: How AI Agents Avoid Getting Rekt

Liquidation is the single greatest threat to a leveraged trading agent. Unlike a human who can top up margin manually, an agent operating autonomously can blow through a position in milliseconds with no oversight. Building liquidation protection is not optional — it is table stakes for any agent running perpetual futures on Purple Flea Trading.

Understanding Perpetual Funding Risk

Perpetual futures have two distinct P&L components that agents often conflate:

1. Mark P&L: Unrealized gain/loss from price movement
2. Funding P&L: Periodic payments between longs and shorts to keep the perpetual pegged to spot

Funding is paid every 8 hours on most perpetual exchanges. During periods of extreme market sentiment, funding rates can reach 0.3-0.5% per 8 hours — that's 1.5% per day, 45% per month drain on a long position during a bull run. An agent holding a leveraged long during high funding must account for this decay.

Most agents focus exclusively on price liquidation. Funding-driven margin erosion is equally dangerous and far less visible.

The Five Layers of Liquidation Protection

Layer 1: Maximum Leverage Caps

Hard-code a maximum leverage per strategy tier. Never allow runtime arguments to override these caps:

MAX_LEVERAGE = {
    'conservative': 2,
    'moderate': 5,
    'aggressive': 10,
    'speculative': 20  # only with explicit circuit breakers
}

def validate_leverage(leverage: int, strategy: str) -> int:
    cap = MAX_LEVERAGE.get(strategy, 5)
    if leverage > cap:
        raise ValueError(
            f'Leverage {leverage}x exceeds cap {cap}x for {strategy} strategy'
        )
    return leverage

Layer 2: Position Sizing by Kelly Criterion

Kelly Criterion provides the theoretically optimal position size given a known edge and win rate. For perpetuals, use a fractional Kelly (25-50%) to account for estimation error:

def kelly_size(win_rate: float, avg_win: float,
               avg_loss: float, account_size: float,
               kelly_fraction: float = 0.25) -> float:
    """
    Calculate position size using fractional Kelly Criterion.
    win_rate: probability of winning (0-1)
    avg_win: average win as fraction of position (e.g. 0.05 = 5%)
    avg_loss: average loss as fraction of position (e.g. 0.03 = 3%)
    """
    if avg_loss == 0:
        return 0.0
    b = avg_win / avg_loss  # reward-to-risk ratio
    p = win_rate
    q = 1 - win_rate
    kelly = (b * p - q) / b
    if kelly <= 0:
        return 0.0  # no edge, don't trade
    fractional_kelly = kelly * kelly_fraction
    return account_size * fractional_kelly

Layer 3: Dynamic Stop-Loss Placement

Static stop-losses get hunted by market makers. Dynamic stops that adapt to volatility (ATR-based) are more robust:

def atr_stop(entry_price: float, atr: float,
              direction: str, multiplier: float = 2.0) -> float:
    """
    Place stop-loss at N * ATR from entry.
    ATR = Average True Range (measures recent volatility).
    """
    stop_distance = atr * multiplier
    if direction == 'long':
        return entry_price - stop_distance
    else:  # short
        return entry_price + stop_distance

def percent_stop(entry_price: float,
                  max_loss_pct: float, direction: str) -> float:
    """Simple percentage-based stop."""
    if direction == 'long':
        return entry_price * (1 - max_loss_pct)
    else:
        return entry_price * (1 + max_loss_pct)

Layer 4: Margin Ratio Monitoring

Poll your margin ratio continuously and reduce position size before reaching the maintenance margin threshold:

import requests

def get_margin_ratio(api_key: str, position_id: str) -> float:
    """Fetch current margin ratio from Purple Flea Trading API."""
    resp = requests.get(
        f'https://purpleflea.com/api/trading/position/{position_id}',
        headers={'Authorization': f'Bearer {api_key}'}
    ).json()
    return resp['margin_ratio']  # e.g. 0.15 = 15%

def margin_action(ratio: float) -> str:
    """Determine action based on current margin ratio."""
    if ratio > 0.5:   return 'safe'
    if ratio > 0.3:   return 'monitor'
    if ratio > 0.2:   return 'reduce_25pct'
    if ratio > 0.15:  return 'reduce_50pct'
    return 'close_immediately'

Layer 5: Funding Rate Circuit Breakers

If funding rate exceeds a threshold, automatically close or flip your position to avoid paying into a crowded trade:

def check_funding(api_key: str, symbol: str) -> dict:
    resp = requests.get(
        f'https://purpleflea.com/api/trading/funding/{symbol}',
        headers={'Authorization': f'Bearer {api_key}'}
    ).json()
    return resp  # {'rate': 0.0003, 'next_payment_in': 14400}

def should_close_for_funding(funding_rate: float,
                               position_direction: str,
                               threshold: float = 0.001) -> bool:
    """
    Close if paying funding above threshold.
    Longs pay when funding > 0; shorts pay when funding < 0.
    """
    if position_direction == 'long' and funding_rate > threshold:
        return True
    if position_direction == 'short' and funding_rate < -threshold:
        return True
    return False

Purple Flea Trading API Circuit Breakers

Purple Flea Trading exposes dedicated circuit breaker endpoints that agents can use to set hard limits at the exchange level — not just in local code. Exchange-level limits persist even if the agent crashes:

# Set exchange-level stop-loss (survives agent crashes)
curl -X POST https://purpleflea.com/api/trading/stop-loss \
  -H 'Authorization: Bearer YOUR_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "position_id": "pos_abc123",
    "stop_price": 42000.00,
    "type": "market"
  }'

# Set take-profit
curl -X POST https://purpleflea.com/api/trading/take-profit \
  -H 'Authorization: Bearer YOUR_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "position_id": "pos_abc123",
    "take_profit_price": 48000.00,
    "type": "limit"
  }'

# Set max daily loss (account-level circuit breaker)
curl -X POST https://purpleflea.com/api/trading/daily-loss-limit \
  -H 'Authorization: Bearer YOUR_API_KEY' \
  -H 'Content-Type: application/json' \
  -d '{
    "max_loss_usdc": 50.00,
    "action": "close_all_and_halt"
  }'

The LiquidationGuard Class

The following Python class wraps all five protection layers into a single guardian that runs alongside your trading agent, monitoring positions and intervening before losses become catastrophic.

import requests
import time
import logging
from dataclasses import dataclass, field
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('LiquidationGuard')

BASE = 'https://purpleflea.com/api/trading'

@dataclass
class Position:
    id: str
    symbol: str
    direction: str  # 'long' or 'short'
    entry_price: float
    size_usdc: float
    leverage: int
    stop_loss: Optional[float] = None
    take_profit: Optional[float] = None
    funding_paid: float = 0.0


class LiquidationGuard:
    """
    Monitors all open positions and enforces liquidation
    protection rules in real time.
    """

    def __init__(self, api_key: str,
                  max_margin_drawdown: float = 0.20,
                  max_funding_rate: float = 0.001,
                  poll_interval: int = 30):
        self.api_key = api_key
        self.max_drawdown = max_margin_drawdown
        self.max_funding = max_funding_rate
        self.poll_interval = poll_interval
        self.positions: dict[str, Position] = {}
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        })
        self.daily_loss = 0.0
        self.daily_loss_limit = 100.0  # USDC

    # --- Registration ---

    def watch(self, position: Position) -> None:
        """Register a position for monitoring."""
        self.positions[position.id] = position
        # Set exchange-level stop immediately
        if position.stop_loss:
            self._set_exchange_stop(position)
        logger.info(f'Watching {position.symbol} {position.id}')

    def unwatch(self, position_id: str) -> None:
        self.positions.pop(position_id, None)

    # --- Exchange-Level Stops ---

    def _set_exchange_stop(self, pos: Position) -> bool:
        try:
            r = self.session.post(f'{BASE}/stop-loss', json={
                'position_id': pos.id,
                'stop_price': pos.stop_loss,
                'type': 'market'
            })
            return r.json().get('success', False)
        except Exception as e:
            logger.error(f'Failed to set stop: {e}')
            return False

    # --- Position Checks ---

    def _get_position_state(self, pos_id: str) -> Optional[dict]:
        try:
            return self.session.get(f'{BASE}/position/{pos_id}').json()
        except Exception as e:
            logger.error(f'Failed to fetch position {pos_id}: {e}')
            return None

    def _get_funding_rate(self, symbol: str) -> float:
        try:
            r = self.session.get(f'{BASE}/funding/{symbol}')
            return r.json().get('rate', 0.0)
        except Exception:
            return 0.0

    def _close_position(self, pos: Position, reason: str) -> bool:
        logger.warning(f'Closing {pos.id} ({pos.symbol}): {reason}')
        try:
            r = self.session.post(f'{BASE}/close', json={
                'position_id': pos.id,
                'type': 'market'
            })
            result = r.json()
            if result.get('success'):
                pnl = result.get('pnl', 0.0)
                self.daily_loss += min(0, pnl)
                self.unwatch(pos.id)
                return True
        except Exception as e:
            logger.error(f'Close failed: {e}')
        return False

    def _reduce_position(self, pos: Position,
                          reduction_pct: float, reason: str) -> bool:
        logger.info(f'Reducing {pos.id} by {reduction_pct:.0%}: {reason}')
        reduce_size = pos.size_usdc * reduction_pct
        try:
            r = self.session.post(f'{BASE}/reduce', json={
                'position_id': pos.id,
                'reduce_by_usdc': reduce_size
            })
            if r.json().get('success'):
                pos.size_usdc -= reduce_size
                return True
        except Exception as e:
            logger.error(f'Reduce failed: {e}')
        return False

    # --- Main Guard Logic ---

    def check_position(self, pos: Position) -> str:
        """Check a single position and take action if needed."""
        state = self._get_position_state(pos.id)
        if not state:
            return 'unknown'

        margin_ratio = state.get('margin_ratio', 1.0)
        mark_price = state.get('mark_price', pos.entry_price)
        unrealized_pnl = state.get('unrealized_pnl', 0.0)

        # 1. Margin ratio checks
        action = self._margin_action(margin_ratio)
        if action == 'close_immediately':
            self._close_position(pos, f'margin_ratio={margin_ratio:.2%}')
            return 'closed'
        elif action == 'reduce_50pct':
            self._reduce_position(pos, 0.5, f'margin_ratio={margin_ratio:.2%}')
            return 'reduced'
        elif action == 'reduce_25pct':
            self._reduce_position(pos, 0.25, f'margin_ratio={margin_ratio:.2%}')
            return 'reduced'

        # 2. Funding rate check
        funding = self._get_funding_rate(pos.symbol)
        if self._should_close_for_funding(funding, pos.direction):
            self._close_position(pos, f'funding_rate={funding:.4%}')
            return 'closed'

        # 3. Daily loss limit
        if abs(self.daily_loss) >= self.daily_loss_limit:
            self._close_position(pos, 'daily_loss_limit_reached')
            return 'closed'

        return action

    def _margin_action(self, ratio: float) -> str:
        if ratio > 0.5:   return 'safe'
        if ratio > 0.3:   return 'monitor'
        if ratio > 0.2:   return 'reduce_25pct'
        if ratio > 0.15:  return 'reduce_50pct'
        return 'close_immediately'

    def _should_close_for_funding(self, rate: float, direction: str) -> bool:
        if direction == 'long' and rate > self.max_funding:
            return True
        if direction == 'short' and rate < -self.max_funding:
            return True
        return False

    # --- Main Loop ---

    def run_forever(self) -> None:
        """Monitor all watched positions continuously."""
        logger.info(f'LiquidationGuard started — watching {len(self.positions)} positions')
        while True:
            for pos_id in list(self.positions.keys()):
                pos = self.positions.get(pos_id)
                if pos:
                    status = self.check_position(pos)
                    if status not in ('safe', 'monitor', 'unknown'):
                        logger.info(f'Action taken on {pos_id}: {status}')
            time.sleep(self.poll_interval)


# --- Example Usage ---

if __name__ == '__main__':
    guard = LiquidationGuard(
        api_key='YOUR_API_KEY',
        max_margin_drawdown=0.20,
        max_funding_rate=0.001,
        poll_interval=30
    )

    # Register your open position
    my_position = Position(
        id='pos_abc123',
        symbol='BTC-PERP',
        direction='long',
        entry_price=65000.0,
        size_usdc=500.0,
        leverage=5,
        stop_loss=59000.0,
        take_profit=75000.0
    )
    guard.watch(my_position)
    guard.run_forever()

Common Liquidation Scenarios and Defenses

Backtesting Your Protection Parameters

Before deploying the LiquidationGuard in production, backtest your stop-loss placement and funding thresholds against historical data. The key metric is MAE (Maximum Adverse Excursion) — how far against you a position moves before reversing:

def backtest_stops(trades: list, stop_multiplier: float) -> dict:
    """
    Evaluate stop-loss hit rate at a given ATR multiplier.
    trades: list of {'entry', 'exit', 'high', 'low', 'atr', 'direction'}
    """
    hits = 0
    survivors = 0
    for t in trades:
        stop = t['entry'] - t['atr'] * stop_multiplier \
               if t['direction'] == 'long' \
               else t['entry'] + t['atr'] * stop_multiplier
        if t['direction'] == 'long' and t['low'] < stop:
            hits += 1
        elif t['direction'] == 'short' and t['high'] > stop:
            hits += 1
        else:
            survivors += 1

    return {
        'stop_multiplier': stop_multiplier,
        'hit_rate': hits / len(trades),
        'survivor_rate': survivors / len(trades),
        'recommended': stop_multiplier if hits / len(trades) < 0.15 else 'increase'
    }

Getting Started

1. Register at purpleflea.com/register
2. Explore Purple Flea Trading with paper positions first
3. Deploy the LiquidationGuard as a separate process before going live
4. Set your daily loss limit via the API circuit breaker endpoint
5. Start with 2x leverage maximum until your guard is battle-tested

Liquidation is not a risk to manage after the fact — it is a design constraint to engineer around from day one. Agents that survive their first bear market are the ones that treated protection as a first-class feature.