Market Microstructure Traditional

#!/usr/bin/env python3
"""Market maker simulation with inventory management and P&L tracking.

Simulates a simple market maker using the Avellaneda-Stoikov framework.
The MM quotes bid/ask prices that skew based on inventory, captures spread,
and faces adverse selection from informed traders. Includes --demo mode.

Usage:
    python scripts/market_maker_sim.py --demo
    python scripts/market_maker_sim.py --demo --gamma 0.1 --n-steps 5000

Dependencies:
    uv pip install numpy pandas matplotlib

Environment Variables:
    None required for --demo mode.
"""

import argparse
import sys
from dataclasses import dataclass, field
from typing import Optional

import numpy as np
import pandas as pd


# ── Data Structures ─────────────────────────────────────────────────


@dataclass
class MarketMakerState:
    """Current state of the market maker."""

    cash: float = 0.0
    inventory: int = 0
    n_trades: int = 0
    n_buys_filled: int = 0
    n_sells_filled: int = 0
    total_spread_captured: float = 0.0
    total_adverse_selection_loss: float = 0.0
    max_inventory: int = 0
    min_inventory: int = 0


@dataclass
class SimulationConfig:
    """Configuration for the market maker simulation."""

    n_steps: int = 2000
    initial_price: float = 100.0
    volatility: float = 0.001
    gamma: float = 0.05           # risk aversion
    k: float = 1.5                # order arrival intensity
    base_spread_bps: float = 10.0 # minimum spread
    max_inventory: int = 50       # inventory limit
    informed_fraction: float = 0.15
    trade_probability: float = 0.3
    seed: int = 42


@dataclass
class TradeRecord:
    """Record of a single fill."""

    step: int
    side: str             # "buy" or "sell" (from MM perspective)
    price: float
    midprice: float
    inventory_before: int
    inventory_after: int
    pnl: float            # immediate mark-to-market P&L
    is_informed: bool


@dataclass
class SimulationResult:
    """Complete simulation output."""

    config: SimulationConfig
    prices: np.ndarray
    inventories: np.ndarray
    pnl_series: np.ndarray
    bid_prices: np.ndarray
    ask_prices: np.ndarray
    trades: list[TradeRecord]
    final_state: MarketMakerState


# ── Price Simulation ────────────────────────────────────────────────


def simulate_price_path(
    n_steps: int,
    initial_price: float,
    volatility: float,
    seed: int = 42,
) -> np.ndarray:
    """Generate a geometric Brownian motion price path.

    Args:
        n_steps: Number of time steps.
        initial_price: Starting price.
        volatility: Per-step return volatility.
        seed: Random seed.

    Returns:
        Array of prices.
    """
    rng = np.random.default_rng(seed)
    returns = rng.normal(0, volatility, n_steps)
    # Add small drift and occasional jumps for realism
    jumps = rng.choice([0, 1], size=n_steps, p=[0.98, 0.02])
    jump_sizes = rng.normal(0, volatility * 5, n_steps) * jumps
    returns += jump_sizes
    prices = initial_price * np.exp(np.cumsum(returns))
    return prices


# ── Avellaneda-Stoikov Quoting ──────────────────────────────────────


def compute_optimal_quotes(
    midprice: float,
    inventory: int,
    gamma: float,
    sigma: float,
    time_remaining: float,
    k: float,
    min_spread_bps: float = 5.0,
) -> tuple[float, float]:
    """Compute optimal bid and ask using Avellaneda-Stoikov model.

    The reservation price shifts away from inventory, and the optimal
    spread depends on volatility and risk aversion.

    Args:
        midprice: Current fair price.
        inventory: Current inventory (positive = long).
        gamma: Risk aversion parameter.
        sigma: Volatility (per step).
        time_remaining: Fraction of session remaining (0 to 1).
        k: Order arrival rate parameter.
        min_spread_bps: Minimum spread floor in basis points.

    Returns:
        Tuple of (bid_price, ask_price).
    """
    # Reservation price: skew away from inventory
    T = max(time_remaining, 0.001)
    reservation_price = midprice - inventory * gamma * (sigma ** 2) * T

    # Optimal spread
    spread = gamma * (sigma ** 2) * T + (2.0 / gamma) * np.log(1 + gamma / k)

    # Apply minimum spread
    min_spread = midprice * min_spread_bps / 10_000
    spread = max(spread, min_spread)

    bid = reservation_price - spread / 2
    ask = reservation_price + spread / 2

    return bid, ask


# ── Simulation Engine ───────────────────────────────────────────────


def run_simulation(config: SimulationConfig) -> SimulationResult:
    """Run a complete market maker simulation.

    Args:
        config: Simulation parameters.

    Returns:
        SimulationResult with full history.
    """
    rng = np.random.default_rng(config.seed)

    # Generate true price path
    true_prices = simulate_price_path(
        config.n_steps, config.initial_price, config.volatility, config.seed
    )

    # State tracking arrays
    inventories = np.zeros(config.n_steps)
    pnl_series = np.zeros(config.n_steps)
    bid_prices = np.zeros(config.n_steps)
    ask_prices = np.zeros(config.n_steps)

    state = MarketMakerState()
    trades: list[TradeRecord] = []

    for t in range(config.n_steps):
        midprice = true_prices[t]
        time_remaining = (config.n_steps - t) / config.n_steps

        # Compute quotes
        bid, ask = compute_optimal_quotes(
            midprice=midprice,
            inventory=state.inventory,
            gamma=config.gamma,
            sigma=config.volatility,
            time_remaining=time_remaining,
            k=config.k,
            min_spread_bps=config.base_spread_bps,
        )

        # Enforce inventory limits by widening quotes
        if state.inventory >= config.max_inventory:
            bid = midprice * 0.99  # very low bid, discourage more buying
        if state.inventory <= -config.max_inventory:
            ask = midprice * 1.01  # very high ask, discourage more selling

        bid_prices[t] = bid
        ask_prices[t] = ask

        # Simulate incoming order
        if rng.random() < config.trade_probability:
            is_informed = rng.random() < config.informed_fraction

            if is_informed:
                # Informed trader knows future direction
                future_idx = min(t + 50, config.n_steps - 1)
                future_price = true_prices[future_idx]
                # Buy if price is going up, sell if going down
                incoming_side = "buy" if future_price > midprice else "sell"
            else:
                incoming_side = "buy" if rng.random() > 0.5 else "sell"

            if incoming_side == "buy" and ask <= midprice * 1.005:
                # Incoming buy lifts our ask -> we sell
                fill_price = ask
                state.cash += fill_price
                state.inventory -= 1
                state.n_sells_filled += 1
                state.n_trades += 1

                spread_captured = fill_price - midprice
                state.total_spread_captured += spread_captured

                trades.append(TradeRecord(
                    step=t,
                    side="sell",
                    price=fill_price,
                    midprice=midprice,
                    inventory_before=state.inventory + 1,
                    inventory_after=state.inventory,
                    pnl=spread_captured,
                    is_informed=is_informed,
                ))

            elif incoming_side == "sell" and bid >= midprice * 0.995:
                # Incoming sell hits our bid -> we buy
                fill_price = bid
                state.cash -= fill_price
                state.inventory += 1
                state.n_buys_filled += 1
                state.n_trades += 1

                spread_captured = midprice - fill_price
                state.total_spread_captured += spread_captured

                trades.append(TradeRecord(
                    step=t,
                    side="buy",
                    price=fill_price,
                    midprice=midprice,
                    inventory_before=state.inventory - 1,
                    inventory_after=state.inventory,
                    pnl=spread_captured,
                    is_informed=is_informed,
                ))

        # Track state
        state.max_inventory = max(state.max_inventory, state.inventory)
        state.min_inventory = min(state.min_inventory, state.inventory)
        inventories[t] = state.inventory
        # Mark-to-market P&L: cash + inventory * midprice
        pnl_series[t] = state.cash + state.inventory * midprice

    # Compute adverse selection loss from informed trades
    informed_trades = [tr for tr in trades if tr.is_informed]
    for tr in informed_trades:
        future_idx = min(tr.step + 50, config.n_steps - 1)
        future_mid = true_prices[future_idx]
        if tr.side == "sell":
            # We sold; if price went up, we lost
            loss = max(0, future_mid - tr.price)
        else:
            # We bought; if price went down, we lost
            loss = max(0, tr.price - future_mid)
        state.total_adverse_selection_loss += loss

    return SimulationResult(
        config=config,
        prices=true_prices,
        inventories=inventories,
        pnl_series=pnl_series,
        bid_prices=bid_prices,
        ask_prices=ask_prices,
        trades=trades,
        final_state=state,
    )


# ── Reporting ───────────────────────────────────────────────────────


def print_report(result: SimulationResult) -> None:
    """Print a summary report of the simulation.

    Args:
        result: Completed simulation result.
    """
    state = result.final_state
    config = result.config
    final_mid = result.prices[-1]
    final_pnl = state.cash + state.inventory * final_mid

    print("=" * 55)
    print("  MARKET MAKER SIMULATION REPORT")
    print("=" * 55)
    print()
    print(f"  Configuration:")
    print(f"    Steps:              {config.n_steps:,}")
    print(f"    Initial price:      ${config.initial_price:.2f}")
    print(f"    Volatility:         {config.volatility:.4f} per step")
    print(f"    Risk aversion (γ):  {config.gamma:.3f}")
    print(f"    Informed fraction:  {config.informed_fraction:.0%}")
    print(f"    Max inventory:      ±{config.max_inventory}")
    print()
    print(f"  Trading Activity:")
    print(f"    Total fills:        {state.n_trades:,}")
    print(f"    Buys (we bought):   {state.n_buys_filled:,}")
    print(f"    Sells (we sold):    {state.n_sells_filled:,}")
    print(f"    Fill rate:          {state.n_trades / config.n_steps:.1%} of steps")
    print()
    print(f"  Inventory:")
    print(f"    Final inventory:    {state.inventory:+d}")
    print(f"    Max inventory:      {state.max_inventory:+d}")
    print(f"    Min inventory:      {state.min_inventory:+d}")
    print()
    print(f"  P&L Breakdown:")
    print(f"    Spread captured:    ${state.total_spread_captured:+.2f}")
    print(f"    Adverse selection:  ${-state.total_adverse_selection_loss:.2f}")
    print(f"    Inventory value:    ${state.inventory * final_mid:+.2f}")
    print(f"    Final M2M P&L:      ${final_pnl:+.2f}")
    print()

    # Per-trade statistics
    if result.trades:
        spreads = [abs(tr.price - tr.midprice) / tr.midprice * 10_000
                   for tr in result.trades]
        print(f"  Per-Trade Stats:")
        print(f"    Avg spread captured: {np.mean(spreads):.2f} bps")
        print(f"    Median:              {np.median(spreads):.2f} bps")
        print(f"    Std:                 {np.std(spreads):.2f} bps")

        informed_count = sum(1 for tr in result.trades if tr.is_informed)
        print(f"    Informed fills:      {informed_count} ({informed_count/len(result.trades)*100:.1f}%)")
    print()

    # Sharpe-like metric
    if len(result.pnl_series) > 1:
        pnl_returns = np.diff(result.pnl_series)
        if np.std(pnl_returns) > 0:
            sharpe = np.mean(pnl_returns) / np.std(pnl_returns) * np.sqrt(252 * 24 * 60)
            print(f"  Risk-Adjusted (annualized Sharpe estimate): {sharpe:.2f}")
        print()


def plot_simulation(
    result: SimulationResult,
    output_path: Optional[str] = None,
) -> None:
    """Plot simulation results in a 4-panel chart.

    Args:
        result: Completed simulation result.
        output_path: If provided, save to file instead of showing.
    """
    try:
        import matplotlib.pyplot as plt
    except ImportError:
        print("matplotlib not installed, skipping plot.")
        return

    fig, axes = plt.subplots(4, 1, figsize=(14, 12), sharex=True)
    steps = np.arange(result.config.n_steps)

    # Panel 1: Price and quotes
    ax1 = axes[0]
    ax1.plot(steps, result.prices, color="black", linewidth=0.5, label="True Price")
    ax1.plot(steps, result.bid_prices, color="blue", linewidth=0.3, alpha=0.5, label="Bid")
    ax1.plot(steps, result.ask_prices, color="red", linewidth=0.3, alpha=0.5, label="Ask")

    buy_trades = [tr for tr in result.trades if tr.side == "buy"]
    sell_trades = [tr for tr in result.trades if tr.side == "sell"]
    if buy_trades:
        ax1.scatter([tr.step for tr in buy_trades],
                    [tr.price for tr in buy_trades],
                    color="green", s=8, alpha=0.6, label="We Buy", zorder=5)
    if sell_trades:
        ax1.scatter([tr.step for tr in sell_trades],
                    [tr.price for tr in sell_trades],
                    color="red", s=8, alpha=0.6, label="We Sell", zorder=5)
    ax1.set_ylabel("Price")
    ax1.set_title("Market Maker Simulation — Price and Quotes")
    ax1.legend(loc="upper left", fontsize=7)

    # Panel 2: Inventory
    ax2 = axes[1]
    ax2.fill_between(steps, 0, result.inventories, alpha=0.4,
                     color="steelblue", label="Inventory")
    ax2.axhline(y=0, color="black", linewidth=0.5)
    ax2.axhline(y=result.config.max_inventory, color="red",
                linewidth=0.5, linestyle="--", label="Limit")
    ax2.axhline(y=-result.config.max_inventory, color="red",
                linewidth=0.5, linestyle="--")
    ax2.set_ylabel("Inventory")
    ax2.set_title("Inventory Over Time")
    ax2.legend(fontsize=7)

    # Panel 3: Mark-to-Market P&L
    ax3 = axes[2]
    ax3.plot(steps, result.pnl_series, color="darkgreen", linewidth=0.8)
    ax3.axhline(y=0, color="black", linewidth=0.5)
    ax3.set_ylabel("P&L ($)")
    ax3.set_title("Cumulative Mark-to-Market P&L")

    # Panel 4: Spread over time
    ax4 = axes[3]
    spread_bps = (result.ask_prices - result.bid_prices) / result.prices * 10_000
    ax4.plot(steps, spread_bps, color="purple", linewidth=0.5)
    ax4.axhline(y=np.mean(spread_bps), color="red", linestyle="--",
                label=f"Mean: {np.mean(spread_bps):.1f} bps")
    ax4.set_ylabel("Spread (bps)")
    ax4.set_xlabel("Time Step")
    ax4.set_title("Quoted Spread Over Time")
    ax4.legend(fontsize=7)

    plt.tight_layout()
    if output_path:
        plt.savefig(output_path, dpi=150)
        print(f"Plot saved to {output_path}")
    else:
        plt.show()


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Run market maker simulation from command line."""
    parser = argparse.ArgumentParser(
        description="Market maker simulation with Avellaneda-Stoikov quoting"
    )
    parser.add_argument(
        "--demo", action="store_true",
        help="Run with synthetic price path (no API key needed)",
    )
    parser.add_argument("--n-steps", type=int, default=2000, help="Simulation steps")
    parser.add_argument("--price", type=float, default=100.0, help="Initial price")
    parser.add_argument("--volatility", type=float, default=0.001, help="Per-step volatility")
    parser.add_argument("--gamma", type=float, default=0.05, help="Risk aversion (0.01-1.0)")
    parser.add_argument("--informed", type=float, default=0.15, help="Informed trader fraction")
    parser.add_argument("--max-inv", type=int, default=50, help="Max inventory limit")
    parser.add_argument("--seed", type=int, default=42, help="Random seed")
    parser.add_argument("--plot", action="store_true", help="Show plots")
    parser.add_argument("--save-plot", type=str, default=None, help="Save plot to file")

    args = parser.parse_args()

    if not args.demo:
        print("Use --demo to run the simulation. Use --help for options.")
        sys.exit(1)

    config = SimulationConfig(
        n_steps=args.n_steps,
        initial_price=args.price,
        volatility=args.volatility,
        gamma=args.gamma,
        informed_fraction=args.informed,
        max_inventory=args.max_inv,
        seed=args.seed,
    )

    print(f"Running market maker simulation ({config.n_steps:,} steps)...")
    print()

    result = run_simulation(config)
    print_report(result)

    # Parameter sensitivity analysis
    print("=== Gamma Sensitivity ===")
    print(f"  {'Gamma':>8}  {'Trades':>8}  {'Final P&L':>12}  {'Avg Spread':>12}")
    print(f"  {'─' * 8}  {'─' * 8}  {'─' * 12}  {'─' * 12}")
    for g in [0.01, 0.05, 0.1, 0.2, 0.5]:
        test_config = SimulationConfig(
            n_steps=config.n_steps,
            initial_price=config.initial_price,
            volatility=config.volatility,
            gamma=g,
            informed_fraction=config.informed_fraction,
            max_inventory=config.max_inventory,
            seed=config.seed,
        )
        test_result = run_simulation(test_config)
        final_pnl = test_result.final_state.cash + test_result.final_state.inventory * test_result.prices[-1]
        avg_spread = np.mean(
            (test_result.ask_prices - test_result.bid_prices) / test_result.prices * 10_000
        )
        print(f"  {g:>8.3f}  {test_result.final_state.n_trades:>8,}  ${final_pnl:>+11.2f}  {avg_spread:>10.1f} bps")
    print()

    if args.plot or args.save_plot:
        plot_simulation(result, output_path=args.save_plot)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Bid-ask spread analysis: compute quoted, effective, and realized spreads.

Analyzes order book and trade data to decompose spreads into adverse selection,
inventory, and order processing components. Includes --demo mode with synthetic
data so no API key is required to run.

Usage:
    python scripts/spread_analysis.py --demo
    python scripts/spread_analysis.py --input trades.csv

Dependencies:
    uv pip install numpy pandas matplotlib

Environment Variables:
    None required for --demo mode.
"""

import argparse
import sys
from dataclasses import dataclass, field
from typing import Optional

import numpy as np
import pandas as pd


# ── Data Structures ─────────────────────────────────────────────────


@dataclass
class SpreadMetrics:
    """Aggregated spread metrics for a trading session."""

    quoted_spread_bps: float
    effective_spread_bps: float
    realized_spread_bps: float
    adverse_selection_bps: float
    price_impact_bps: float
    n_trades: int

    def summary(self) -> str:
        """Return a human-readable summary."""
        lines = [
            "=== Spread Analysis Summary ===",
            f"  Trades analyzed:        {self.n_trades:,}",
            f"  Quoted spread:          {self.quoted_spread_bps:.2f} bps",
            f"  Effective spread:       {self.effective_spread_bps:.2f} bps",
            f"  Realized spread (5s):   {self.realized_spread_bps:.2f} bps",
            f"  Adverse selection:      {self.adverse_selection_bps:.2f} bps",
            f"  Avg price impact:       {self.price_impact_bps:.2f} bps",
        ]
        return "\n".join(lines)


# ── Synthetic Data Generation ───────────────────────────────────────


def generate_synthetic_orderbook(
    n_snapshots: int = 1000,
    base_price: float = 100.0,
    base_spread_bps: float = 10.0,
    volatility: float = 0.001,
    seed: int = 42,
) -> pd.DataFrame:
    """Generate synthetic order book snapshots with bid/ask prices.

    Args:
        n_snapshots: Number of time steps.
        base_price: Starting midprice.
        base_spread_bps: Average quoted spread in basis points.
        volatility: Per-step return volatility.
        seed: Random seed for reproducibility.

    Returns:
        DataFrame with columns: timestamp, bid, ask, midprice.
    """
    rng = np.random.default_rng(seed)

    # Random walk midprice
    returns = rng.normal(0, volatility, n_snapshots)
    midprices = base_price * np.exp(np.cumsum(returns))

    # Spread varies around base with some noise
    half_spread_frac = (base_spread_bps / 10_000) / 2
    spread_noise = rng.uniform(0.5, 1.5, n_snapshots)

    bids = midprices * (1 - half_spread_frac * spread_noise)
    asks = midprices * (1 + half_spread_frac * spread_noise)

    timestamps = pd.date_range("2025-01-01", periods=n_snapshots, freq="100ms")

    return pd.DataFrame({
        "timestamp": timestamps,
        "bid": bids,
        "ask": asks,
        "midprice": midprices,
    })


def generate_synthetic_trades(
    orderbook: pd.DataFrame,
    trades_per_snapshot: float = 0.3,
    informed_fraction: float = 0.15,
    seed: int = 123,
) -> pd.DataFrame:
    """Generate synthetic trades against an order book.

    Informed trades tend to buy before price rises and sell before drops.
    Uninformed trades are random.

    Args:
        orderbook: Order book snapshots from generate_synthetic_orderbook.
        trades_per_snapshot: Average number of trades per snapshot.
        informed_fraction: Fraction of trades that are informed.
        seed: Random seed for reproducibility.

    Returns:
        DataFrame with columns: timestamp, price, side, midprice, is_informed.
    """
    rng = np.random.default_rng(seed)
    trades = []

    midprices = orderbook["midprice"].values
    future_returns = np.zeros(len(midprices))
    lookahead = 50  # look 50 steps ahead for "information"
    for i in range(len(midprices) - lookahead):
        future_returns[i] = (midprices[i + lookahead] - midprices[i]) / midprices[i]

    for i, row in orderbook.iterrows():
        if rng.random() > trades_per_snapshot:
            continue

        is_informed = rng.random() < informed_fraction

        if is_informed:
            # Informed traders buy before price rises, sell before drops
            side = "buy" if future_returns[i] > 0 else "sell"
        else:
            side = "buy" if rng.random() > 0.5 else "sell"

        # Trade executes at ask for buys, bid for sells, with some noise
        if side == "buy":
            price = row["ask"] * (1 + rng.uniform(0, 0.0001))
        else:
            price = row["bid"] * (1 - rng.uniform(0, 0.0001))

        trades.append({
            "timestamp": row["timestamp"],
            "price": price,
            "side": side,
            "midprice": row["midprice"],
            "is_informed": is_informed,
        })

    return pd.DataFrame(trades)


# ── Spread Computation ──────────────────────────────────────────────


def compute_quoted_spread(orderbook: pd.DataFrame) -> pd.Series:
    """Compute quoted spread in basis points for each snapshot.

    Args:
        orderbook: DataFrame with bid and ask columns.

    Returns:
        Series of quoted spreads in basis points.
    """
    midprice = (orderbook["bid"] + orderbook["ask"]) / 2
    return (orderbook["ask"] - orderbook["bid"]) / midprice * 10_000


def compute_effective_spread(trades: pd.DataFrame) -> pd.Series:
    """Compute effective half-spread in basis points for each trade.

    The effective spread measures the actual cost paid relative to the
    midprice at the time of the trade.

    Args:
        trades: DataFrame with price, side, and midprice columns.

    Returns:
        Series of effective half-spreads in basis points.
    """
    sign = trades["side"].map({"buy": 1.0, "sell": -1.0})
    return sign * (trades["price"] - trades["midprice"]) / trades["midprice"] * 10_000


def compute_realized_spread(
    trades: pd.DataFrame,
    orderbook: pd.DataFrame,
    delay_steps: int = 50,
) -> pd.Series:
    """Compute realized spread: what the market maker actually keeps.

    Realized spread = trade_sign * (trade_price - midprice_after_delay).
    The difference between effective and realized spread is adverse selection.

    Args:
        trades: Trade DataFrame with timestamp, price, side columns.
        orderbook: Order book DataFrame with timestamp and midprice.
        delay_steps: Number of order book snapshots to look ahead.

    Returns:
        Series of realized half-spreads in basis points.
    """
    # Map each trade to the midprice delay_steps later
    ob_midprices = orderbook.set_index("timestamp")["midprice"]
    ob_timestamps = orderbook["timestamp"].values

    realized = []
    for _, trade in trades.iterrows():
        # Find the orderbook index closest to trade timestamp
        idx = np.searchsorted(ob_timestamps, trade["timestamp"])
        future_idx = min(idx + delay_steps, len(ob_midprices) - 1)
        future_mid = ob_midprices.iloc[future_idx]

        sign = 1.0 if trade["side"] == "buy" else -1.0
        rs = sign * (trade["price"] - future_mid) / trade["midprice"] * 10_000
        realized.append(rs)

    return pd.Series(realized, index=trades.index)


def compute_kyle_lambda(
    trades: pd.DataFrame,
    interval: str = "1s",
) -> dict:
    """Estimate Kyle's lambda from trade data.

    Aggregates signed volume and midprice changes over intervals, then
    regresses price changes on signed volume.

    Args:
        trades: Trade DataFrame with timestamp, price, side, midprice.
        interval: Time interval for aggregation.

    Returns:
        Dict with lambda estimate and R-squared.
    """
    df = trades.copy()
    df["signed_volume"] = df["side"].map({"buy": 1.0, "sell": -1.0})
    df = df.set_index("timestamp")

    agg = df.resample(interval).agg({
        "signed_volume": "sum",
        "midprice": "last",
    }).dropna()

    agg["midprice_change"] = agg["midprice"].diff()
    agg = agg.dropna()

    if len(agg) < 10:
        return {"lambda": 0.0, "r_squared": 0.0, "n_intervals": len(agg)}

    X = agg["signed_volume"].values.reshape(-1, 1)
    y = agg["midprice_change"].values

    X_with_const = np.column_stack([np.ones(len(X)), X])
    beta, residuals, _, _ = np.linalg.lstsq(X_with_const, y, rcond=None)

    y_hat = X_with_const @ beta
    ss_res = np.sum((y - y_hat) ** 2)
    ss_tot = np.sum((y - y.mean()) ** 2)
    r_sq = 1 - ss_res / ss_tot if ss_tot > 0 else 0.0

    return {
        "lambda": float(beta[1]),
        "r_squared": float(r_sq),
        "n_intervals": len(agg),
    }


def analyze_spreads(
    trades: pd.DataFrame,
    orderbook: pd.DataFrame,
    delay_steps: int = 50,
) -> SpreadMetrics:
    """Run full spread analysis and return aggregated metrics.

    Args:
        trades: Trade DataFrame.
        orderbook: Order book DataFrame.
        delay_steps: Steps for realized spread delay.

    Returns:
        SpreadMetrics with all computed values.
    """
    quoted = compute_quoted_spread(orderbook)
    effective = compute_effective_spread(trades)
    realized = compute_realized_spread(trades, orderbook, delay_steps)

    adverse_selection = effective.mean() - realized.mean()

    # Price impact: midprice move in trade direction after delay
    signs = trades["side"].map({"buy": 1.0, "sell": -1.0})
    ob_midprices = orderbook.set_index("timestamp")["midprice"]
    ob_timestamps = orderbook["timestamp"].values

    impacts = []
    for _, trade in trades.iterrows():
        idx = np.searchsorted(ob_timestamps, trade["timestamp"])
        future_idx = min(idx + delay_steps, len(ob_midprices) - 1)
        future_mid = ob_midprices.iloc[future_idx]
        sign = 1.0 if trade["side"] == "buy" else -1.0
        impact = sign * (future_mid - trade["midprice"]) / trade["midprice"] * 10_000
        impacts.append(impact)

    return SpreadMetrics(
        quoted_spread_bps=float(quoted.mean()),
        effective_spread_bps=float(effective.mean()),
        realized_spread_bps=float(realized.mean()),
        adverse_selection_bps=float(adverse_selection),
        price_impact_bps=float(np.mean(impacts)),
        n_trades=len(trades),
    )


# ── Visualization ───────────────────────────────────────────────────


def plot_spread_analysis(
    orderbook: pd.DataFrame,
    trades: pd.DataFrame,
    metrics: SpreadMetrics,
    output_path: Optional[str] = None,
) -> None:
    """Plot spread analysis results.

    Args:
        orderbook: Order book DataFrame.
        trades: Trade DataFrame.
        metrics: Computed spread metrics.
        output_path: If provided, save plot to this path instead of showing.
    """
    try:
        import matplotlib.pyplot as plt
    except ImportError:
        print("matplotlib not installed, skipping plot.")
        return

    fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True)

    # Panel 1: Price with bid/ask
    ax1 = axes[0]
    ax1.fill_between(orderbook["timestamp"], orderbook["bid"], orderbook["ask"],
                     alpha=0.3, color="blue", label="Bid-Ask")
    ax1.plot(orderbook["timestamp"], orderbook["midprice"],
             color="black", linewidth=0.5, label="Midprice")

    buys = trades[trades["side"] == "buy"]
    sells = trades[trades["side"] == "sell"]
    ax1.scatter(buys["timestamp"], buys["price"], color="green",
                s=10, alpha=0.6, label="Buys", zorder=5)
    ax1.scatter(sells["timestamp"], sells["price"], color="red",
                s=10, alpha=0.6, label="Sells", zorder=5)
    ax1.set_ylabel("Price")
    ax1.set_title("Order Book and Trades")
    ax1.legend(loc="upper left", fontsize=8)

    # Panel 2: Quoted spread over time
    ax2 = axes[1]
    quoted = compute_quoted_spread(orderbook)
    ax2.plot(orderbook["timestamp"], quoted, color="purple", linewidth=0.5)
    ax2.axhline(y=metrics.quoted_spread_bps, color="red", linestyle="--",
                label=f"Mean: {metrics.quoted_spread_bps:.1f} bps")
    ax2.set_ylabel("Quoted Spread (bps)")
    ax2.set_title("Quoted Spread Over Time")
    ax2.legend(fontsize=8)

    # Panel 3: Spread decomposition bar chart
    ax3 = axes[2]
    components = ["Quoted", "Effective", "Realized", "Adverse\nSelection"]
    values = [
        metrics.quoted_spread_bps,
        metrics.effective_spread_bps,
        metrics.realized_spread_bps,
        metrics.adverse_selection_bps,
    ]
    colors = ["steelblue", "seagreen", "coral", "crimson"]
    ax3.bar(components, values, color=colors, edgecolor="black", linewidth=0.5)
    for i, v in enumerate(values):
        ax3.text(i, v + 0.2, f"{v:.2f}", ha="center", fontsize=9)
    ax3.set_ylabel("Basis Points")
    ax3.set_title("Spread Decomposition")

    plt.tight_layout()
    if output_path:
        plt.savefig(output_path, dpi=150)
        print(f"Plot saved to {output_path}")
    else:
        plt.show()


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Run spread analysis from command line."""
    parser = argparse.ArgumentParser(
        description="Bid-ask spread analysis and decomposition"
    )
    parser.add_argument(
        "--demo", action="store_true",
        help="Run with synthetic data (no API key needed)",
    )
    parser.add_argument(
        "--input", type=str, default=None,
        help="Path to CSV with trade data (columns: timestamp, price, side, midprice)",
    )
    parser.add_argument(
        "--plot", action="store_true",
        help="Show spread analysis plots",
    )
    parser.add_argument(
        "--save-plot", type=str, default=None,
        help="Save plot to file instead of displaying",
    )
    args = parser.parse_args()

    if args.demo:
        print("Generating synthetic order book and trades...")
        orderbook = generate_synthetic_orderbook(
            n_snapshots=2000,
            base_price=150.0,
            base_spread_bps=12.0,
            volatility=0.0008,
        )
        trades = generate_synthetic_trades(
            orderbook,
            trades_per_snapshot=0.25,
            informed_fraction=0.20,
        )
        print(f"  Order book snapshots: {len(orderbook):,}")
        print(f"  Trades generated:     {len(trades):,}")
        print()

    elif args.input:
        trades = pd.read_csv(args.input, parse_dates=["timestamp"])
        if "midprice" not in trades.columns:
            print("Error: CSV must have a 'midprice' column.")
            sys.exit(1)
        # Build a minimal orderbook from trade data
        orderbook = pd.DataFrame({
            "timestamp": trades["timestamp"],
            "midprice": trades["midprice"],
            "bid": trades["midprice"] * 0.9995,
            "ask": trades["midprice"] * 1.0005,
        })
        print(f"Loaded {len(trades):,} trades from {args.input}")
        print()
    else:
        print("Specify --demo or --input. Use --help for options.")
        sys.exit(1)

    # Run analysis
    print("Computing spread metrics...")
    metrics = analyze_spreads(trades, orderbook, delay_steps=50)
    print()
    print(metrics.summary())
    print()

    # Kyle's lambda
    print("Estimating Kyle's lambda...")
    kyle = compute_kyle_lambda(trades, interval="1s")
    print(f"  Lambda:       {kyle['lambda']:.6f}")
    print(f"  R-squared:    {kyle['r_squared']:.4f}")
    print(f"  N intervals:  {kyle['n_intervals']}")
    print()

    # Informed vs uninformed breakdown (demo only)
    if args.demo and "is_informed" in trades.columns:
        informed = trades[trades["is_informed"]]
        uninformed = trades[~trades["is_informed"]]
        print("=== Informed vs Uninformed Breakdown ===")
        print(f"  Informed trades:   {len(informed):,} ({len(informed)/len(trades)*100:.1f}%)")
        print(f"  Uninformed trades: {len(uninformed):,} ({len(uninformed)/len(trades)*100:.1f}%)")

        eff_informed = compute_effective_spread(informed)
        eff_uninformed = compute_effective_spread(uninformed)
        print(f"  Effective spread (informed):   {eff_informed.mean():.2f} bps")
        print(f"  Effective spread (uninformed): {eff_uninformed.mean():.2f} bps")
        print()

    # Plot
    if args.plot or args.save_plot:
        plot_spread_analysis(orderbook, trades, metrics, output_path=args.save_plot)


if __name__ == "__main__":
    main()