Flash Crash Recovery for AI Agents: Detection, Circuit Breakers, and Opportunistic Buying

Flash crashes are violent, chaotic, and over in minutes — but for AI agents running the right playbook, they are also the most profitable trading opportunities of the year. The difference between getting liquidated and making a killing comes down to three things: detection speed, disciplined circuit breakers, and pre-loaded opportunistic buy orders. This is the definitive guide.

What Causes Flash Crashes?

A flash crash is a rapid, deep decline in asset prices that recovers substantially within minutes to hours, distinguished from a regular crash by its velocity (typically >5% in under 5 minutes) and by the disproportionate role of automated selling.

Flash crashes in crypto markets are caused by a confluence of factors that reinforce each other in a cascade. Understanding the mechanics is the first step to surviving — and profiting from — them.

Liquidation Cascades

The most common flash crash trigger in crypto is a liquidation cascade. Here's the anatomy:

Other Flash Crash Causes

• Exchange technical failures — A major exchange experiencing downtime concentrates sell flow on remaining venues, temporarily overwhelming their order books
• Stablecoin de-peg events — UST/LUNA in May 2022 caused cascading crypto-wide crashes as traders unwound positions to cover stablecoin losses
• Macro shock headline — Regulatory announcements, central bank decisions, or geopolitical events can trigger instantaneous algorithmic sell programs
• DeFi liquidations — On-chain collateral liquidations from protocols like Aave and Compound create their own concentrated sell pressure during volatile periods
• Thin holiday markets — Flash crashes occur disproportionately during low-liquidity periods (weekends, Asian holidays) when fewer market makers are active

Flash Crash Detection Signals

Early detection is the entire game. An agent that identifies a flash crash 30 seconds before the bottom can lock in protective positions and queue opportunistic buys at prices that will recover. An agent that detects it 30 seconds after the bottom is already too late to fully benefit. Here are the three primary detection signals:

Signal 1: Price Velocity

Price velocity measures the rate of price change over a rolling window. Normal BTC/USD price velocity is 0.1-0.5% per minute. Flash crash velocity exceeds 2-5% per minute and can reach 10%+ during the most extreme events.

from collections import deque
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class VelocityDetector:
    """
    Measures price velocity over multiple windows simultaneously.
    Shorter windows = faster detection but more false positives.
    Longer windows = slower but more reliable signals.
    """
    asset: str
    windows: tuple = (60, 300, 900)  # 1min, 5min, 15min in seconds
    alert_thresholds: tuple = (3.0, 5.0, 8.0)  # %/window for each window
    price_buffer: deque = field(default_factory=lambda: deque(maxlen=1000))

    def update(self, price: float) -> Optional[dict]:
        """Add new price tick. Returns alert if velocity threshold exceeded."""
        now = time.time()
        self.price_buffer.append((now, price))

        alerts = []
        for window_sec, threshold_pct in zip(self.windows, self.alert_thresholds):
            window_start = now - window_sec
            window_prices = [
                (ts, p) for ts, p in self.price_buffer
                if ts >= window_start
            ]
            if len(window_prices) < 2:
                continue

            oldest_price = window_prices[0][1]
            change_pct = (price - oldest_price) / oldest_price * 100

            if abs(change_pct) >= threshold_pct:
                alerts.append({
                    "window_sec": window_sec,
                    "change_pct": round(change_pct, 3),
                    "direction": "DOWN" if change_pct < 0 else "UP",
                    "threshold": threshold_pct
                })

        if alerts:
            return {
                "asset": self.asset,
                "current_price": price,
                "timestamp": now,
                "velocity_alerts": alerts,
                "severity": self._compute_severity(alerts)
            }
        return None

    def _compute_severity(self, alerts: list) -> str:
        """Classify severity based on how many windows trigger simultaneously."""
        if len(alerts) >= 3:
            return "CRITICAL"  # All windows triggered = severe flash crash
        elif len(alerts) == 2:
            return "HIGH"
        else:
            return "MEDIUM"

Signal 2: Volume Spikes

Flash crashes are almost always accompanied by volume that is 5-20x the baseline. The combination of abnormal volume and rapid price decline is a much stronger signal than either alone. Pure price velocity can be triggered by a thin-book move; volume confirms that significant selling activity is occurring.

import statistics

class VolumeAnomalyDetector:
    def __init__(self, lookback_bars: int = 20, spike_multiplier: float = 5.0):
        self.lookback_bars = lookback_bars
        self.spike_multiplier = spike_multiplier
        self.volume_history: deque = deque(maxlen=lookback_bars)

    def update(self, volume: float) -> dict:
        """
        Check if current bar volume is anomalous relative to recent history.
        Returns detection result with z-score and multiplier.
        """
        result = {
            "volume": volume,
            "is_anomaly": False,
            "multiplier": None,
            "z_score": None
        }

        if len(self.volume_history) >= 5:
            hist = list(self.volume_history)
            avg_volume = statistics.mean(hist)
            std_volume = statistics.stdev(hist) if len(hist) > 1 else 1

            multiplier = volume / max(avg_volume, 0.001)
            z_score = (volume - avg_volume) / max(std_volume, 0.001)

            result["multiplier"] = round(multiplier, 2)
            result["z_score"] = round(z_score, 2)
            result["avg_volume"] = round(avg_volume, 2)
            result["is_anomaly"] = multiplier >= self.spike_multiplier or z_score >= 4.0

        self.volume_history.append(volume)
        return result

Signal 3: Bid-Ask Spread Widening

When market makers pull their quotes during a crash, bid-ask spreads widen dramatically. This is both a detection signal (crash is happening) and a risk signal (execution costs just exploded). Agents should monitor spread as a percentage of mid-price:

• Normal conditions: BTC spread 0.01-0.03% of mid
• Elevated stress: 0.05-0.15% — slow down trading, widen limit order offsets
• Flash crash conditions: 0.3-2.0%+ — halt market orders, use limit orders only
• Crisis: >2% — pause all trading, protect existing positions only

class SpreadMonitor:
    def __init__(self, baseline_spread_pct: float = 0.02):
        self.baseline = baseline_spread_pct
        self.spread_history: deque = deque(maxlen=100)

    def update(self, bid: float, ask: float) -> dict:
        mid = (bid + ask) / 2
        spread_pct = (ask - bid) / mid * 100
        multiplier = spread_pct / self.baseline

        self.spread_history.append(spread_pct)

        # Classify market conditions
        if spread_pct < 0.05:
            condition = "NORMAL"
        elif spread_pct < 0.15:
            condition = "ELEVATED"
        elif spread_pct < 0.5:
            condition = "STRESSED"
        elif spread_pct < 2.0:
            condition = "FLASH_CRASH"
        else:
            condition = "CRISIS"

        return {
            "bid": bid, "ask": ask, "mid": round(mid, 2),
            "spread_pct": round(spread_pct, 4),
            "spread_multiplier": round(multiplier, 2),
            "condition": condition
        }

Composite Flash Crash Detector

The strongest signal comes from combining all three detectors. A composite score that requires multiple signals to agree before triggering dramatically reduces false positives:

from enum import Enum

class AlertLevel(Enum):
    NONE = 0
    WATCH = 1     # 1 signal triggered — monitor closely
    WARNING = 2  # 2 signals — reduce new position opens
    FLASH = 3    # All 3 signals — flash crash protocol active

class CompositeFlashCrashDetector:
    def __init__(self, asset: str):
        self.asset = asset
        self.velocity = VelocityDetector(asset=asset)
        self.volume = VolumeAnomalyDetector(spike_multiplier=4.0)
        self.spread = SpreadMonitor(baseline_spread_pct=0.02)
        self.current_level = AlertLevel.NONE
        self.level_history: deque = deque(maxlen=50)

    def update(
        self,
        price: float,
        volume: float,
        bid: float,
        ask: float
    ) -> dict:
        """Process new market data tick. Returns composite alert."""
        vel_alert = self.velocity.update(price)
        vol_result = self.volume.update(volume)
        spread_result = self.spread.update(bid, ask)

        signals_triggered = sum([
            vel_alert is not None and vel_alert["velocity_alerts"][0]["direction"] == "DOWN",
            vol_result["is_anomaly"],
            spread_result["condition"] in ("STRESSED", "FLASH_CRASH", "CRISIS")
        ])

        if signals_triggered >= 3:
            new_level = AlertLevel.FLASH
        elif signals_triggered == 2:
            new_level = AlertLevel.WARNING
        elif signals_triggered == 1:
            new_level = AlertLevel.WATCH
        else:
            new_level = AlertLevel.NONE

        level_changed = new_level != self.current_level
        self.current_level = new_level
        self.level_history.append((time.time(), new_level))

        return {
            "asset": self.asset,
            "price": price,
            "alert_level": new_level.name,
            "alert_value": new_level.value,
            "level_changed": level_changed,
            "signals_triggered": signals_triggered,
            "velocity": vel_alert,
            "volume": vol_result,
            "spread": spread_result
        }

Circuit Breaker Implementation

Once a flash crash is detected, the first priority is protection. This is handled by circuit breakers — automated risk controls that halt or limit trading activity when market conditions become unsafe. The most important principle: circuit breakers must be faster than human reaction time to be useful.

Tiered Circuit Breaker Levels

import asyncio
from enum import Enum
from dataclasses import dataclass, field
import httpx

class BreakerLevel(Enum):
    NORMAL = 0
    WATCH = 1
    WARNING = 2
    FLASH = 3
    KILL = 4

class TradingCircuitBreaker:
    def __init__(
        self,
        agent_id: str,
        purple_flea_token: str,
        daily_loss_limit_usd: float = 500.0,
        position_loss_limit_pct: float = 0.05  # 5% of position value
    ):
        self.agent_id = agent_id
        self.token = purple_flea_token
        self.daily_loss_limit = daily_loss_limit_usd
        self.position_loss_limit_pct = position_loss_limit_pct
        self.current_level = BreakerLevel.NORMAL
        self.daily_pnl = 0.0
        self.tripped_at = None
        self.min_flash_recovery_secs = 600  # 10 min minimum halt

    async def cancel_all_orders(self):
        """Cancel all open orders via Purple Flea Trading API."""
        async with httpx.AsyncClient() as client:
            resp = await client.delete(
                "https://purpleflea.com/api/trading/orders",
                headers={"Authorization": f"Bearer {self.token}"},
                params={"agent_id": self.agent_id, "all": "true"}
            )
            print(f"[CB] All orders cancelled: {resp.status_code}")

    async def close_all_positions(self):
        """Emergency close all positions (Kill Switch only)."""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://purpleflea.com/api/trading/close-all",
                headers={"Authorization": f"Bearer {self.token}"},
                json={"agent_id": self.agent_id, "order_type": "market"}
            )
            print(f"[CB] KILL SWITCH: All positions closed: {resp.status_code}")

    async def trip(self, level: BreakerLevel, reason: str):
        """Elevate circuit breaker to specified level and take protective action."""
        if level.value <= self.current_level.value:
            return  # Already at equal or higher level

        self.current_level = level
        self.tripped_at = time.time()
        print(f"[CB] TRIPPED to {level.name}: {reason}")

        if level == BreakerLevel.WARNING:
            await self.cancel_all_orders()

        elif level == BreakerLevel.FLASH:
            await self.cancel_all_orders()
            # Arm opportunistic buy queue (described below)
            print("[CB] Opportunistic buy queue armed for recovery phase")

        elif level == BreakerLevel.KILL:
            await self.cancel_all_orders()
            await self.close_all_positions()
            print("[CB] KILL SWITCH ACTIVATED — manual reset required")

    def can_trade(self) -> tuple[bool, str]:
        """Check if trading is currently permitted."""
        if self.current_level == BreakerLevel.KILL:
            return False, "Kill switch active — manual reset required"
        if self.current_level == BreakerLevel.FLASH:
            elapsed = time.time() - (self.tripped_at or 0)
            if elapsed < self.min_flash_recovery_secs:
                remaining = self.min_flash_recovery_secs - elapsed
                return False, f"Flash halt: {remaining:.0f}s remaining"
        if self.current_level == BreakerLevel.WARNING:
            return False, "WARNING: limit orders only, reduced size"
        return True, "OK"

    async def update_pnl(self, daily_pnl: float):
        """Called on each position update. Trips kill switch on loss limit breach."""
        self.daily_pnl = daily_pnl
        if daily_pnl <= -self.daily_loss_limit:
            await self.trip(
                BreakerLevel.KILL,
                f"Daily loss limit hit: ${daily_pnl:.2f} (limit: ${-self.daily_loss_limit:.2f})"
            )

Opportunistic Buying During Crashes

After the protective layer is in place, prepared agents switch to offense. Flash crashes are rare situations where price collapses far below any reasonable estimate of fair value — and the recovery is statistically predictable.

Pre-Loading Opportunistic Buy Orders

The key insight is to pre-load buy orders before the crash, at levels that would only be reached during a genuine flash crash. This avoids the problem of trying to place orders into a chaotic market with widened spreads and delayed execution.

from dataclasses import dataclass
from typing import List
import asyncio
import httpx

@dataclass
class OpportunisticBuyOrder:
    asset: str
    limit_price: float
    quantity_usd: float
    crash_depth_pct: float      # How far below current price
    max_hold_hours: int = 48   # Auto-cancel if not filled within N hours
    take_profit_pct: float = 5.0  # TP at +5% from fill price
    stop_loss_pct: float = 3.0   # SL at -3% from fill price

class FlashCrashOpportunityEngine:
    def __init__(self, agent_id: str, token: str, reserve_pct: float = 0.25):
        self.agent_id = agent_id
        self.token = token
        self.reserve_pct = reserve_pct  # Keep 25% of capital for crash buying
        self.active_crash_orders: List[str] = []  # Order IDs

    def compute_crash_buy_levels(self, current_price: float, asset: str) -> List[OpportunisticBuyOrder]:
        """
        Generate a ladder of buy orders at increasing crash depths.
        Uses a tranche system: bigger discount = bigger position.
        """
        tranches = [
            (0.10, 0.25),  # 10% crash → 25% of crash budget
            (0.20, 0.35),  # 20% crash → 35% of crash budget
            (0.35, 0.40),  # 35% crash → 40% of crash budget (biggest dislocation)
        ]

        crash_budget_usd = 1000  # Example: $1000 reserved for crash buying
        orders = []

        for depth_pct, size_pct in tranches:
            limit_price = current_price * (1 - depth_pct)
            quantity_usd = crash_budget_usd * size_pct

            orders.append(OpportunisticBuyOrder(
                asset=asset,
                limit_price=round(limit_price, 2),
                quantity_usd=quantity_usd,
                crash_depth_pct=depth_pct * 100,
                max_hold_hours=72,
                take_profit_pct=8.0,   # Expect partial recovery
                stop_loss_pct=5.0
            ))

        return orders

    async def arm_crash_orders(self, current_price: float, asset: str = "BTC-USD"):
        """Submit the crash buy ladder to Purple Flea Trading API."""
        orders = self.compute_crash_buy_levels(current_price, asset)

        async with httpx.AsyncClient() as client:
            for order in orders:
                print(f"Arming crash buy: {asset} @ ${order.limit_price:,.2f} "
                      f"(${order.quantity_usd:.0f}, -{order.crash_depth_pct:.0f}% depth)")
                # Submit limit buy order via Purple Flea API
                resp = await client.post(
                    "https://purpleflea.com/api/trading/orders",
                    headers={"Authorization": f"Bearer {self.token}"},
                    json={
                        "agent_id": self.agent_id,
                        "market": asset,
                        "side": "buy",
                        "type": "limit",
                        "price": order.limit_price,
                        "size_usd": order.quantity_usd,
                        "reduce_only": False,
                        "tif": "gtc",  # Good till cancelled
                        "tp_price": round(order.limit_price * (1 + order.take_profit_pct/100), 2),
                        "sl_price": round(order.limit_price * (1 - order.stop_loss_pct/100), 2)
                    }
                )
                if resp.status_code == 200:
                    order_id = resp.json().get("order_id")
                    self.active_crash_orders.append(order_id)
                    print(f"  Order submitted: {order_id}")

Recovery Pattern Recognition

Not all crashes recover equally. Recognizing the recovery pattern in the first few minutes helps agents decide whether to hold crash buys or exit quickly:

Recovery Confirmation Signals

Before scaling into crash buys aggressively, wait for at least two of these confirmation signals:

• Volume normalizes: Volume drops back below 3x the baseline after the crash spike
• Spread tightens: Bid-ask spread returns below 0.1% — market makers are back
• Price velocity turns positive: 5-minute velocity flips from negative to positive
• Order book depth restores: Bid-side depth recovers to >50% of pre-crash levels
• Funding rate normalizes: Perpetual funding rate, which spikes negative during sell cascades, returns toward zero

Purple Flea Trading API Kill Switches

Purple Flea's Trading API, powered by Hyperliquid's perpetual exchange engine, provides 275 markets and several built-in risk management endpoints that agents can use as programmatic kill switches:

• DELETE /api/trading/orders?all=true — Cancel all open orders across all markets instantly
• POST /api/trading/close-all — Close all open positions at market price
• POST /api/trading/risk/pause — Pause all automated order submission for the agent
• PUT /api/trading/risk/limits — Dynamically adjust position size limits and daily loss thresholds

Complete Flash Crash Monitor Agent

The following is a production-ready agent that integrates all components: multi-signal detection, tiered circuit breakers, and an opportunistic buy queue:

#!/usr/bin/env python3
"""
Flash Crash Monitor and Recovery Agent
Integrates with Purple Flea Trading API
"""
import asyncio
import httpx
import time
from datetime import datetime

PURPLE_FLEA_BASE = "https://purpleflea.com/api"
AGENT_TOKEN = "pf_live_your_token_here"
AGENT_ID = "agent_flash_001"
MONITOR_ASSET = "BTC-USD"
DAILY_LOSS_LIMIT_USD = 500.0

class FlashCrashAgent:
    def __init__(self):
        self.detector = CompositeFlashCrashDetector(MONITOR_ASSET)
        self.breaker = TradingCircuitBreaker(
            agent_id=AGENT_ID,
            purple_flea_token=AGENT_TOKEN,
            daily_loss_limit_usd=DAILY_LOSS_LIMIT_USD
        )
        self.opportunity_engine = FlashCrashOpportunityEngine(
            agent_id=AGENT_ID,
            token=AGENT_TOKEN,
            reserve_pct=0.25
        )
        self.last_known_price = None
        self.crash_buy_orders_armed = False
        self.stats = {
            "ticks_processed": 0,
            "flash_events": 0,
            "crash_buys_filled": 0,
            "total_pnl_usd": 0.0
        }

    async def fetch_market_data(self) -> dict:
        """Fetch current market data from Purple Flea API."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{PURPLE_FLEA_BASE}/trading/market/{MONITOR_ASSET}",
                headers={"Authorization": f"Bearer {AGENT_TOKEN}"},
                timeout=5.0
            )
            return resp.json()

    async def process_tick(self, market_data: dict):
        """Main tick processing logic."""
        price = market_data["mark_price"]
        bid = market_data["best_bid"]
        ask = market_data["best_ask"]
        volume = market_data.get("volume_1m", 0)

        self.stats["ticks_processed"] += 1

        # Run composite detection
        alert = self.detector.update(price=price, volume=volume, bid=bid, ask=ask)

        # Map alert to circuit breaker level
        alert_value = alert["alert_value"]
        if alert_value >= 3:
            await self.breaker.trip(BreakerLevel.FLASH, "All 3 flash crash signals active")
            if not self.crash_buy_orders_armed:
                await self.opportunity_engine.arm_crash_orders(
                    current_price=self.last_known_price or price,
                    asset=MONITOR_ASSET
                )
                self.crash_buy_orders_armed = True
                self.stats["flash_events"] += 1

        elif alert_value == 2:
            await self.breaker.trip(BreakerLevel.WARNING, "2 flash crash signals active")

        elif alert_value == 1:
            await self.breaker.trip(BreakerLevel.WATCH, "1 flash crash signal active")

        else:
            # Market normal — reset crash order armed flag
            if self.crash_buy_orders_armed:
                print(f"[{datetime.now().strftime('%H:%M:%S')}] Market normalizing — monitoring crash order fills")
                self.crash_buy_orders_armed = False

        self.last_known_price = price

        # Check daily P&L
        daily_pnl = market_data.get("daily_pnl", 0.0)
        await self.breaker.update_pnl(daily_pnl)

        # Log status
        can_trade, reason = self.breaker.can_trade()
        status_icon = "✓" if can_trade else "✗"
        print(
            f"[{datetime.now().strftime('%H:%M:%S')}] {MONITOR_ASSET} ${price:,.2f} | "
            f"Alert: {alert['alert_level']} | Trade: {status_icon} {reason}"
        )

    async def run(self, poll_interval_ms: int = 1000):
        print(f"Flash Crash Monitor started — tracking {MONITOR_ASSET}")
        print(f"Daily loss limit: ${DAILY_LOSS_LIMIT_USD:,.2f}")
        while True:
            try:
                data = await self.fetch_market_data()
                await self.process_tick(data)
            except httpx.RequestError as e:
                print(f"Network error: {e}")
            except Exception as e:
                print(f"Unexpected error: {e}")
                # Unknown errors trigger FLASH level defensively
                await self.breaker.trip(BreakerLevel.FLASH, f"Unexpected error: {e}")
            await asyncio.sleep(poll_interval_ms / 1000)

if __name__ == "__main__":
    agent = FlashCrashAgent()
    asyncio.run(agent.run(poll_interval_ms=500))  # Poll every 500ms

Historical Flash Crash Analysis

Understanding historical flash crashes informs better thresholds and strategies:

Position Sizing During Crash Buying

Proper position sizing during crash buying is as important as the entry trigger. The Kelly Criterion adapted for binary-outcome events (crash recovery vs. continued decline) can guide crash buy sizing:

def kelly_crash_size(
    win_probability: float,     # Historical recovery probability (e.g., 0.75)
    win_return: float,           # Expected gain if recovery (e.g., 0.15 = 15%)
    loss_return: float,          # Expected loss if no recovery (e.g., -0.10 = -10%)
    kelly_fraction: float = 0.25  # Use 25% Kelly (fractional Kelly for safety)
) -> float:
    """
    Returns the fraction of available capital to deploy in crash buying.
    Kelly formula: f = (p * b - q) / b where b = win/loss ratio
    """
    loss_probability = 1 - win_probability
    b = win_return / abs(loss_return)  # Odds ratio
    kelly_full = (win_probability * b - loss_probability) / b
    fractional_kelly = max(0, kelly_full * kelly_fraction)  # Never negative

    return {
        "full_kelly_pct": round(kelly_full * 100, 2),
        "fractional_kelly_pct": round(fractional_kelly * 100, 2),
        "suggested_max_pct_of_capital": round(fractional_kelly * 100, 2)
    }

# Example: BTC 20% crash with 75% historical recovery probability
result = kelly_crash_size(
    win_probability=0.75,   # 75% chance of recovery based on history
    win_return=0.12,        # 12% gain (partial recovery captured)
    loss_return=-0.08,      # 8% loss if price continues down
    kelly_fraction=0.25     # Use 25% Kelly for safety
)
print(result)
# {'full_kelly_pct': 35.42, 'fractional_kelly_pct': 8.85, 'suggested_max_pct_of_capital': 8.85}
# → Deploy max 8.85% of total capital per crash event

Conclusion

Flash crashes are one of the most extreme tests of any trading agent's risk infrastructure. The agents that survive — and profit — are those that built their systems for this scenario before it happened. The core playbook:

1. Monitor all three signals: Price velocity, volume spikes, and spread widening must all be tracked in real-time
2. Require multi-signal confirmation: A single signal is noise; three signals is a flash crash
3. Pre-load protection: Circuit breakers must be coded, tested, and verified before markets open
4. Pre-load opportunity: Crash buy limit orders should be placed at −10%, −20%, and −35% levels as standing orders
5. Respect the kill switch: When daily loss limits are hit, stop — recovering from a kill switch is far easier than recovering from a blown account
6. Check for structural breaks: Before buying a crash, verify it's a liquidity event, not a fundamental collapse
7. Size with Kelly: Even in the best crash-buying scenario, never deploy more than 10-15% of capital on any single event

Purple Flea's Trading API provides 275 perpetual markets with cancel-all and close-all endpoints that serve as the kill switch backbone for any serious agent trading system. Combined with the faucet for risk-free testing and the multi-chain wallet for capital management, the full infrastructure for crash-resilient agent trading is available today.