High-Frequency Trading and Latency Arbitrage › Market Impact and Flash Crashes

def simulate_flash_crash(
    n_periods: int = 200,
    n_market_makers: int = 10,
    initial_price: float = 100.0,
    normal_volatility: float = 0.05,
    shock_magnitude: float = 2.0,  # Standard deviations
    mm_withdrawal_threshold: float = 1.5,  # Volatility multiple to withdraw
    mm_risk_limit: float = 500,  # Inventory limit before withdrawal
):
    """
    Simulate a flash crash scenario with HFT market maker withdrawal.

    Parameters:
    -----------
    n_periods : Number of time periods
    n_market_makers : Number of HFT market makers
    initial_price : Starting price
    normal_volatility : Normal volatility per period (%)
    shock_magnitude : Size of initial shock in std devs
    mm_withdrawal_threshold : Volatility level triggering withdrawal
    mm_risk_limit : Inventory level triggering withdrawal
    """
    np.random.seed(42)

    prices = [initial_price]
    active_mms = [n_market_makers]
    total_liquidity = [
        n_market_makers * 1000
    ]  # Shares available at best bid/ask

    # Market maker states
    mm_inventories = np.zeros(n_market_makers)
    mm_active = np.ones(n_market_makers, dtype=bool)

    recent_vol = normal_volatility

    for t in range(1, n_periods):
        # Large sell order at t=50
        if t == 50:
            shock = -shock_magnitude * normal_volatility
        else:
            shock = 0

        # Normal random price movement
        random_move = np.random.normal(0, normal_volatility)

        # Price impact depends on available liquidity
        liquidity_factor = total_liquidity[-1] / (n_market_makers * 1000)
        price_impact = 1 / max(
            liquidity_factor, 0.1
        )  # Higher impact with less liquidity

        price_change = (random_move + shock) * price_impact
        new_price = prices[-1] * (1 + price_change / 100)

        # Update recent volatility
        if t > 5:
            recent_returns = np.diff(np.log(prices[-5:])) * 100
            recent_vol = np.std(recent_returns)

        # Market makers update their state
        for i in range(n_market_makers):
            if mm_active[i]:
                # Accumulate inventory from the shock
                if t == 50:
                    mm_inventories[i] += np.random.exponential(200)

                # Check withdrawal conditions
                vol_exceeded = (
                    recent_vol > mm_withdrawal_threshold * normal_volatility
                )
                inventory_exceeded = abs(mm_inventories[i]) > mm_risk_limit

                if vol_exceeded or inventory_exceeded:
                    mm_active[i] = False

        # Update liquidity
        current_liquidity = mm_active.sum() * 1000

        # If liquidity gets very low, allow some recovery
        if t > 100 and recent_vol < normal_volatility * 1.2:
            for i in range(n_market_makers):
                if not mm_active[i] and np.random.random() < 0.1:
                    mm_active[i] = True
                    mm_inventories[i] = 0

        prices.append(new_price)
        active_mms.append(mm_active.sum())
        total_liquidity.append(current_liquidity)

    return {
        "prices": np.array(prices),
        "active_mms": np.array(active_mms),
        "total_liquidity": np.array(total_liquidity),
        "min_price": min(prices),
        "max_drawdown": (initial_price - min(prices)) / initial_price * 100,
    }


crash = simulate_flash_crash()