Stop Flash Crashes from Destroying Your Trading System in 90 Seconds

# Personal note: Learned this after the 2010 flash crash cost me real money
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta

@dataclass
class LiquidityMetrics:
    """Real-time liquidity health indicators"""
    timestamp: datetime
    symbol: str
    bid_ask_spread_pct: float
    book_depth_ratio: float  # Current vs 5-min average
    trade_velocity: float     # Trades per second
    price_impact_estimate: float
    liquidity_score: float    # 0-100, <30 is dangerous
    
class LiquidityMonitor:
    def __init__(self, lookback_seconds=300):
        self.lookback = lookback_seconds
        self.spread_history = {}
        self.depth_history = {}
        self.trade_history = {}
        
    def calculate_metrics(self, symbol: str, orderbook: dict, 
                         recent_trades: list) -> LiquidityMetrics:
        """
        Calculate liquidity health metrics
        
        Watch out: Don't use bid_size/ask_size alone - they can be 
        spoofed. Use actual trade velocity as confirmation.
        """
        now = datetime.now()
        
        # Spread analysis
        best_bid = orderbook['bids'][0][0]
        best_ask = orderbook['asks'][0][0]
        mid_price = (best_bid + best_ask) / 2
        spread_pct = ((best_ask - best_bid) / mid_price) * 100
        
        # Book depth (top 5 levels)
        bid_depth = sum([level[1] for level in orderbook['bids'][:5]])
        ask_depth = sum([level[1] for level in orderbook['asks'][:5]])
        total_depth = bid_depth + ask_depth
        
        # Compare to historical average
        if symbol not in self.depth_history:
            self.depth_history[symbol] = []
        self.depth_history[symbol].append({
            'timestamp': now,
            'depth': total_depth
        })
        
        # Clean old data
        cutoff = now - timedelta(seconds=self.lookback)
        self.depth_history[symbol] = [
            d for d in self.depth_history[symbol] 
            if d['timestamp'] > cutoff
        ]
        
        avg_depth = np.mean([d['depth'] for d in self.depth_history[symbol]])
        depth_ratio = total_depth / avg_depth if avg_depth > 0 else 1.0
        
        # Trade velocity
        recent_cutoff = now - timedelta(seconds=10)
        recent_count = len([t for t in recent_trades if t['timestamp'] > recent_cutoff])
        trade_velocity = recent_count / 10.0
        
        # Price impact estimation (conservative)
        # How much would a $10k order move the market?
        test_order_size = 10000 / mid_price
        cumulative_shares = 0
        weighted_price = 0
        
        for price, size in orderbook['asks'][:10]:
            if cumulative_shares >= test_order_size:
                break
            fill_size = min(size, test_order_size - cumulative_shares)
            weighted_price += price * fill_size
            cumulative_shares += fill_size
            
        avg_fill_price = weighted_price / cumulative_shares if cumulative_shares > 0 else mid_price
        price_impact_pct = ((avg_fill_price - mid_price) / mid_price) * 100
        
        # Composite liquidity score (0-100)
        # Lower is worse
        spread_score = max(0, 100 - (spread_pct * 20))  # 5% spread = 0 score
        depth_score = min(100, depth_ratio * 50)         # 2x depth = 100
        velocity_score = min(100, trade_velocity * 10)   # 10 trades/sec = 100
        impact_score = max(0, 100 - (price_impact_pct * 10))  # 10% impact = 0
        
        liquidity_score = (spread_score * 0.3 + 
                          depth_score * 0.3 + 
                          velocity_score * 0.2 + 
                          impact_score * 0.2)
        
        return LiquidityMetrics(
            timestamp=now,
            symbol=symbol,
            bid_ask_spread_pct=spread_pct,
            book_depth_ratio=depth_ratio,
            trade_velocity=trade_velocity,
            price_impact_estimate=price_impact_pct,
            liquidity_score=liquidity_score
        )

from enum import Enum
from typing import Callable, Optional

class ExecutionTier(Enum):
    """Execution strategy tiers based on liquidity"""
    NORMAL = "normal"           # Score 70-100
    CAUTIOUS = "cautious"       # Score 40-70
    DEFENSIVE = "defensive"     # Score 20-40
    EMERGENCY = "emergency"     # Score <20
    HALT = "halt"              # Manual override

@dataclass
class FallbackConfig:
    """Configuration for each execution tier"""
    tier: ExecutionTier
    max_order_size_usd: float
    max_spread_pct: float
    order_type: str  # 'market', 'limit', 'halt'
    limit_offset_bps: int  # Basis points from mid
    max_clip_size_pct: float  # % of book depth
    timeout_seconds: int
    
class FallbackExecutor:
    def __init__(self):
        # Personal note: These thresholds saved me during the Sept 2024 CPI print
        self.configs = {
            ExecutionTier.NORMAL: FallbackConfig(
                tier=ExecutionTier.NORMAL,
                max_order_size_usd=50000,
                max_spread_pct=0.5,
                order_type='limit',
                limit_offset_bps=10,  # 0.1% from mid
                max_clip_size_pct=0.25,
                timeout_seconds=30
            ),
            ExecutionTier.CAUTIOUS: FallbackConfig(
                tier=ExecutionTier.CAUTIOUS,
                max_order_size_usd=10000,
                max_spread_pct=1.0,
                order_type='limit',
                limit_offset_bps=25,
                max_clip_size_pct=0.15,
                timeout_seconds=60
            ),
            ExecutionTier.DEFENSIVE: FallbackConfig(
                tier=ExecutionTier.DEFENSIVE,
                max_order_size_usd=2000,
                max_spread_pct=2.0,
                order_type='limit',
                limit_offset_bps=50,
                max_clip_size_pct=0.05,
                timeout_seconds=120
            ),
            ExecutionTier.EMERGENCY: FallbackConfig(
                tier=ExecutionTier.EMERGENCY,
                max_order_size_usd=0,  # No new orders
                max_spread_pct=999,
                order_type='halt',
                limit_offset_bps=0,
                max_clip_size_pct=0,
                timeout_seconds=0
            )
        }
        
        self.current_tier = ExecutionTier.NORMAL
        self.tier_history = []
        
    def determine_tier(self, metrics: LiquidityMetrics) -> ExecutionTier:
        """
        Select execution tier based on liquidity score
        
        Watch out: Use hysteresis to prevent oscillation between tiers.
        Once you go defensive, require score >50 to go back to cautious.
        """
        score = metrics.liquidity_score
        
        # Tier thresholds with hysteresis
        if score >= 70:
            return ExecutionTier.NORMAL
        elif score >= 40:
            # If already defensive, need >50 to move up
            if self.current_tier == ExecutionTier.DEFENSIVE and score < 50:
                return ExecutionTier.DEFENSIVE
            return ExecutionTier.CAUTIOUS
        elif score >= 20:
            return ExecutionTier.DEFENSIVE
        else:
            return ExecutionTier.EMERGENCY
            
    def execute_order(self, symbol: str, side: str, size_usd: float,
                     metrics: LiquidityMetrics, 
                     orderbook: dict) -> Optional[dict]:
        """
        Execute order with appropriate fallback strategy
        
        Returns: Order details or None if halted
        """
        tier = self.determine_tier(metrics)
        config = self.configs[tier]
        
        # Log tier changes
        if tier != self.current_tier:
            print(f"[{metrics.timestamp}] Tier change: {self.current_tier.value} -> {tier.value}")
            print(f"  Liquidity score: {metrics.liquidity_score:.1f}")
            print(f"  Spread: {metrics.bid_ask_spread_pct:.2f}%")
            self.tier_history.append({
                'timestamp': metrics.timestamp,
                'from_tier': self.current_tier,
                'to_tier': tier,
                'metrics': metrics
            })
            self.current_tier = tier
        
        # Emergency halt
        if tier == ExecutionTier.EMERGENCY:
            print(f"[EMERGENCY HALT] Liquidity score {metrics.liquidity_score:.1f}")
            return None
            
        # Check spread threshold
        if metrics.bid_ask_spread_pct > config.max_spread_pct:
            print(f"[SKIP] Spread {metrics.bid_ask_spread_pct:.2f}% exceeds {config.max_spread_pct}%")
            return None
            
        # Size limits
        order_size = min(size_usd, config.max_order_size_usd)
        
        # Calculate limit price
        mid_price = (orderbook['bids'][0][0] + orderbook['asks'][0][0]) / 2
        offset_pct = config.limit_offset_bps / 10000
        
        if side == 'buy':
            limit_price = mid_price * (1 + offset_pct)
        else:
            limit_price = mid_price * (1 - offset_pct)
            
        # Clip to book depth
        if side == 'buy':
            available_depth = sum([level[1] for level in orderbook['asks'][:5]])
        else:
            available_depth = sum([level[1] for level in orderbook['bids'][:5]])
            
        max_clip_shares = available_depth * config.max_clip_size_pct
        max_clip_usd = max_clip_shares * mid_price
        order_size = min(order_size, max_clip_usd)
        
        return {
            'symbol': symbol,
            'side': side,
            'size_usd': order_size,
            'limit_price': limit_price,
            'tier': tier.value,
            'timeout': config.timeout_seconds,
            'timestamp': metrics.timestamp
        }

from collections import deque
from threading import Lock

class CircuitBreaker:
    def __init__(self):
        self.is_tripped = False
        self.trip_reasons = []
        self.lock = Lock()
        
        # Tracking windows
        self.loss_window = deque(maxlen=100)  # Last 100 trades
        self.fill_quality = deque(maxlen=50)   # Last 50 fills
        
        # Thresholds - adjust based on your risk tolerance
        self.max_drawdown_pct = 5.0
        self.max_consecutive_losses = 8
        self.min_fill_quality_pct = 80.0  # % filled within 10bps of limit
        
    def check_and_trip(self, trade_result: dict, 
                       account_snapshot: dict) -> bool:
        """
        Evaluate circuit breaker conditions
        
        Returns: True if breaker tripped
        """
        with self.lock:
            # Check 1: Drawdown from session high
            current_equity = account_snapshot['equity']
            session_high = account_snapshot.get('session_high', current_equity)
            drawdown_pct = ((session_high - current_equity) / session_high) * 100
            
            if drawdown_pct >= self.max_drawdown_pct:
                self._trip(f"Drawdown {drawdown_pct:.2f}% exceeds {self.max_drawdown_pct}%")
                return True
                
            # Check 2: Consecutive losses
            self.loss_window.append(trade_result.get('pnl', 0) < 0)
            recent_losses = sum(list(self.loss_window)[-self.max_consecutive_losses:])
            
            if recent_losses >= self.max_consecutive_losses:
                self._trip(f"{recent_losses} consecutive losses")
                return True
                
            # Check 3: Fill quality degradation
            if 'expected_price' in trade_result and 'fill_price' in trade_result:
                expected = trade_result['expected_price']
                filled = trade_result['fill_price']
                slippage_bps = abs((filled - expected) / expected) * 10000
                
                quality = 100 if slippage_bps <= 10 else max(0, 100 - slippage_bps)
                self.fill_quality.append(quality)
                
                avg_quality = sum(self.fill_quality) / len(self.fill_quality)
                if avg_quality < self.min_fill_quality_pct:
                    self._trip(f"Fill quality {avg_quality:.1f}% below {self.min_fill_quality_pct}%")
                    return True
                    
            return self.is_tripped
            
    def _trip(self, reason: str):
        """Trip the breaker with reason"""
        if not self.is_tripped:
            self.is_tripped = True
            self.trip_reasons.append({
                'timestamp': datetime.now(),
                'reason': reason
            })
            print(f"\n{'='*60}")
            print(f"CIRCUIT BREAKER TRIPPED: {reason}")
            print(f"{'='*60}\n")
            
    def reset(self, manual_override: bool = False):
        """Reset breaker - requires manual confirmation"""
        if not manual_override:
            raise ValueError("Circuit breaker requires manual reset with manual_override=True")
            
        with self.lock:
            self.is_tripped = False
            print(f"[{datetime.now()}] Circuit breaker manually reset")

import logging
from typing import Dict
import json

class FlashCrashProtectedExecutor:
    def __init__(self, config_path: str = "executor_config.json"):
        # Setup
        self.monitor = LiquidityMonitor(lookback_seconds=300)
        self.fallback = FallbackExecutor()
        self.breaker = CircuitBreaker()
        
        # Logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('execution.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
        # Performance tracking
        self.stats = {
            'orders_submitted': 0,
            'orders_skipped': 0,
            'emergency_halts': 0,
            'avg_liquidity_score': 0,
            'tier_distribution': {tier.value: 0 for tier in ExecutionTier}
        }
        
    def process_order(self, symbol: str, side: str, size_usd: float,
                     orderbook: dict, recent_trades: list,
                     account_snapshot: dict) -> Optional[dict]:
        """
        Main execution path with all protections
        """
        # Step 1: Check circuit breaker
        if self.breaker.is_tripped:
            self.logger.warning("Circuit breaker is tripped - rejecting order")
            return None
            
        # Step 2: Calculate liquidity metrics
        metrics = self.monitor.calculate_metrics(symbol, orderbook, recent_trades)
        self.logger.info(f"Liquidity score: {metrics.liquidity_score:.1f} | "
                        f"Spread: {metrics.bid_ask_spread_pct:.3f}%")
        
        # Update stats
        self.stats['avg_liquidity_score'] = (
            (self.stats['avg_liquidity_score'] * self.stats['orders_submitted'] + 
             metrics.liquidity_score) / (self.stats['orders_submitted'] + 1)
        )
        
        # Step 3: Execute with fallback
        order = self.fallback.execute_order(symbol, side, size_usd, metrics, orderbook)
        
        if order is None:
            self.stats['orders_skipped'] += 1
            if metrics.liquidity_score < 20:
                self.stats['emergency_halts'] += 1
            return None
            
        # Step 4: Track tier
        self.stats['orders_submitted'] += 1
        self.stats['tier_distribution'][order['tier']] += 1
        
        self.logger.info(f"Order submitted: {order['side']} {order['size_usd']:.0f} "
                        f"@ {order['limit_price']:.2f} [{order['tier']}]")
        
        return order
        
    def post_execution_check(self, trade_result: dict, account_snapshot: dict):
        """Check circuit breaker after execution"""
        if self.breaker.check_and_trip(trade_result, account_snapshot):
            self.logger.critical("Circuit breaker tripped - all trading halted")
            # Send alerts here (email, SMS, PagerDuty, etc.)
            
    def get_stats(self) -> Dict:
        """Return performance statistics"""
        return {
            **self.stats,
            'skip_rate_pct': (self.stats['orders_skipped'] / 
                             max(1, self.stats['orders_submitted'] + 
                             self.stats['orders_skipped'])) * 100,
            'current_tier': self.fallback.current_tier.value,
            'breaker_status': 'TRIPPED' if self.breaker.is_tripped else 'ACTIVE'
        }