betex 0.6.0

//! Two-runner order book with implied matching.
//!
//! For 2-runner markets, we can automatically cross-match orders without
//! needing an external hedging service. This is because:
//!   BACK A @ odds `a` ≡ LAY B @ odds `a/(a-1)`
//!
//! This book handles the implied matching internally.

use super::common::*;
use super::error::BookError;
use super::protocol::{
    command::{Command, CommandKind, MarketState, Persistence, Side, TimeInForce},
    reject::RejectReason,
};
use crate::book::common::types::BookOrderInfo;
use crate::types::*;
use chrono::Utc;
use std::collections::{BTreeMap, BTreeSet};
use tracing::error;

use super::common::fast::{OrderKey, OrderStore, RunnerBook};
use serde::{Deserialize, Serialize};

type TickIndex = u16;
type RunnerIdx = usize;
type RunnerLevelKey = (RunnerIdx, Side, TickIndex);
type RestingLevelOrder = (DateTime, OrderId, OrderKey, Money);
type RestingOrdersByLevel = std::collections::HashMap<RunnerLevelKey, Vec<RestingLevelOrder>>;

/// Scratch buffers for command processing.
///
/// ## Memory Reuse Pattern
///
/// The `Scratch` struct provides pre-allocated buffers that are reused across
/// command invocations to avoid repeated heap allocations on the hot path.
///
/// ### Usage Pattern
/// ```ignore
/// // Take ownership of scratch at start of command
/// let mut scratch = std::mem::take(&mut self.scratch);
///
/// // Use scratch buffers during processing
/// scratch.events.clear();
/// scratch.fills.clear();
/// // ... process command ...
///
/// // Return scratch to self at end (even on error paths)
/// self.scratch = scratch;
/// ```
///
/// ### Why This Pattern?
/// - **Avoids allocation on hot path**: Vectors grow to accommodate peak load,
///   then reuse that capacity for subsequent commands.
/// - **Deterministic memory**: No allocator pressure during steady-state operation.
/// - **Thread-local**: Each book owns its scratch, no synchronization needed.
///
/// ### Buffers
/// - `events`: Collects `BookEventEnvelope`s to emit
/// - `fills`: Planned order fills before committing
/// - `matchable`: Orders eligible for matching at current price
/// - `maker_matched_delta`: Tracks cumulative fills per maker during matching
#[derive(Debug, Default)]
struct Scratch {
    maker_matched_delta: std::collections::HashMap<OrderId, i64>,
    matchable: Vec<MatchableOrder>,
    fills: Vec<PlannedFill>,
    events: EventVec,
}

#[derive(Debug, Clone, Copy)]
struct PlannedFill {
    maker_order_id: OrderId,
    maker_side: Side,
    trade_price: OddsX10000,
    fill_amount: Money,
}

#[derive(Debug, Clone)]
struct TradeMatch {
    ts: DateTime,
    trade_id: TradeId,
    maker_order_id: OrderId,
    taker_order_id: OrderId,
    taker_runner_id: RunnerId,
    price: OddsX10000,
    stake: Money,
}

#[derive(Debug, Clone)]
struct MatchRequest {
    ts: DateTime,
    taker_order_id: OrderId,
    taker_account_id: AccountId,
    taker_side: Side,
    taker_runner_id: RunnerId,
    taker_price: OddsX10000,
    taker_stake: Money,
    taker_runner_idx: RunnerIdx,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct TwoRunnerBookSnapshot {
    market_id: MarketId,
    market_kind: MarketKind,
    state: BookMarketState,
    state_before_suspend: Option<BookMarketState>,
    state_before_halt: Option<BookMarketState>,
    next_trade_id: u64,
    next_order_id: u64,
    /// Exactly 2 runners stored as [runner_a, runner_b]
    runner_ids: [RunnerId; 2],
    removed_runners: BTreeSet<RunnerId>,
    close_process: Option<CloseProcessState>,
    orders: OrderStore,
    trades: BTreeMap<TradeId, BookTrade>,
    runner_matched_volume: [Money; 2],
}

/// Two-runner order book with implied matching.
///
/// For 2-runner markets (e.g., tennis, head-to-head), matching is done
/// both directly (BACK A vs LAY A) and through implied odds (BACK A vs BACK B
/// at derived prices).
#[derive(Debug)]
pub struct TwoRunnerBook {
    market_id: MarketId,
    market_kind: MarketKind,
    state: BookMarketState,
    state_before_suspend: Option<BookMarketState>,
    state_before_halt: Option<BookMarketState>,
    next_trade_id: u64,
    next_order_id: u64,
    /// Exactly 2 runners stored as [runner_a, runner_b]
    runner_ids: [RunnerId; 2],
    removed_runners: BTreeSet<RunnerId>,
    runner_books: [RunnerBook; 2],
    close_process: Option<CloseProcessState>,
    orders: OrderStore,
    trades: BTreeMap<TradeId, BookTrade>,
    runner_matched_volume: [Money; 2],

    scratch: Scratch,
}

impl Serialize for TwoRunnerBook {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        TwoRunnerBookSnapshot {
            market_id: self.market_id,
            market_kind: self.market_kind,
            state: self.state,
            state_before_suspend: self.state_before_suspend,
            state_before_halt: self.state_before_halt,
            next_trade_id: self.next_trade_id,
            next_order_id: self.next_order_id,
            runner_ids: self.runner_ids,
            removed_runners: self.removed_runners.clone(),
            close_process: self.close_process,
            orders: self.orders.clone(),
            trades: self.trades.clone(),
            runner_matched_volume: self.runner_matched_volume,
        }
        .serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for TwoRunnerBook {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let snap = TwoRunnerBookSnapshot::deserialize(deserializer)?;
        let mut book = Self {
            market_id: snap.market_id,
            market_kind: snap.market_kind,
            state: snap.state,
            state_before_suspend: snap.state_before_suspend,
            state_before_halt: snap.state_before_halt,
            next_trade_id: snap.next_trade_id,
            next_order_id: snap.next_order_id,
            runner_ids: snap.runner_ids,
            removed_runners: snap.removed_runners,
            runner_books: [RunnerBook::new(), RunnerBook::new()],
            close_process: snap.close_process,
            orders: snap.orders,
            trades: snap.trades,
            runner_matched_volume: snap.runner_matched_volume,
            scratch: Scratch::default(),
        };
        // Deserialization rebuilds runner ladders from the stored order store.
        // This is O(n log n) due to per-level sorting, but bounded by expected snapshot sizes.
        book.rebuild_ladders();
        if let Some((max_oid, _, _)) = book.orders.iter_keys_sorted().last() {
            book.next_order_id = book.next_order_id.max(max_oid.0.saturating_add(1));
        }
        Ok(book)
    }
}

/// Represents a matchable order with its effective price for priority ordering.
#[derive(Debug, Clone)]
struct MatchableOrder {
    order_id: OrderId,
    /// The effective price for matching priority.
    /// For direct matches: the actual order price.
    /// For implied matches: the derived price.
    effective_price: OddsX10000,
    /// Used for FIFO within the same effective price.
    created_at: DateTime,
}

impl Clone for TwoRunnerBook {
    fn clone(&self) -> Self {
        Self {
            market_id: self.market_id,
            market_kind: self.market_kind,
            state: self.state,
            state_before_suspend: self.state_before_suspend,
            state_before_halt: self.state_before_halt,
            next_trade_id: self.next_trade_id,
            next_order_id: self.next_order_id,
            runner_ids: self.runner_ids,
            removed_runners: self.removed_runners.clone(),
            runner_books: self.runner_books.clone(),
            close_process: self.close_process,
            orders: self.orders.clone(),
            trades: self.trades.clone(),
            runner_matched_volume: self.runner_matched_volume,
            scratch: Scratch::default(),
        }
    }
}

impl TwoRunnerBook {
    // Close process batching constants live in `common::close_process`.
    /// Create a new two-runner book.
    ///
    /// # Panics
    /// Panics if the two runner IDs are the same.
    pub fn new(
        market_id: MarketId,
        market_kind: MarketKind,
        runner_a: RunnerId,
        runner_b: RunnerId,
    ) -> Self {
        Self::new_with_capacity(market_id, market_kind, runner_a, runner_b, 20_000)
    }

    pub fn new_with_capacity(
        market_id: MarketId,
        market_kind: MarketKind,
        runner_a: RunnerId,
        runner_b: RunnerId,
        order_store_capacity: usize,
    ) -> Self {
        assert_ne!(
            runner_a, runner_b,
            "Two-runner book requires distinct runners"
        );
        Self {
            market_id,
            market_kind,
            state: BookMarketState::Open,
            state_before_suspend: None,
            state_before_halt: None,
            next_trade_id: 1,
            next_order_id: 1,
            runner_ids: [runner_a, runner_b],
            removed_runners: BTreeSet::new(),
            runner_books: [RunnerBook::new(), RunnerBook::new()],
            close_process: None,
            orders: OrderStore::with_capacity(order_store_capacity),
            trades: BTreeMap::new(),
            runner_matched_volume: [Money::zero(), Money::zero()],
            scratch: Scratch::default(),
        }
    }

    // ========================================================================
    // Odds Derivation
    // ========================================================================

    /// Compute the derived (implied) odds for the opposite runner.
    ///
    /// For odds `o`, the derived odds for the opposite outcome is `o / (o - 1)`.
    /// This is stored as OddsX10000 (multiply by 10000).
    /// The result is snapped to the nearest valid tick on the ladder.
    ///
    /// Example: 2.0 (20000) → 2.0 / 1.0 = 2.0 (20000)
    /// Example: 3.0 (30000) → 3.0 / 2.0 = 1.5 (15000)
    /// Example: 1.5 (15000) → 1.5 / 0.5 = 3.0 (30000)
    fn derived_odds(odds: OddsX10000) -> OddsX10000 {
        // o / (o - 1) = o * 10000 / (o - 10000)
        // We compute: (odds.0 * 10000) / (odds.0 - 10000)
        let o = odds.0 as u64;
        if o <= 10000 {
            // Edge case: odds <= 1.0 shouldn't happen with valid odds
            // Return max valid tick
            return OddsX10000(OddsX10000::MAX);
        }
        let derived = (o * 10000) / (o - 10000);
        let raw = OddsX10000(derived.min(u32::MAX as u64) as u32);

        // Snap to nearest valid tick (floor for conservative matching)
        raw.floor_tick().unwrap_or(OddsX10000(OddsX10000::MAX))
    }

    /// Get the runner index (0 or 1) for a runner ID.
    fn runner_index(&self, runner_id: RunnerId) -> Option<usize> {
        if runner_id == self.runner_ids[0] {
            Some(0)
        } else if runner_id == self.runner_ids[1] {
            Some(1)
        } else {
            None
        }
    }

    /// Get the opposite runner index.
    fn opposite_index(idx: usize) -> usize {
        1 - idx
    }

    fn rebuild_ladders(&mut self) {
        self.runner_books = [RunnerBook::new(), RunnerBook::new()];

        let mut per_level: RestingOrdersByLevel = std::collections::HashMap::new();

        for (oid, key, order) in self.orders.iter_keys_sorted() {
            let is_live = matches!(
                order.info.state,
                BookOrderState::ExecutableUnmatched | BookOrderState::ExecutablePartiallyMatched
            );
            if !is_live {
                continue;
            }
            if self.removed_runners.contains(&order.runner_id) {
                continue;
            }
            let remaining = order.remaining();
            if remaining.0 <= 0 {
                continue;
            }
            let Some(idx) = self.runner_index(order.runner_id) else {
                continue;
            };
            let tick: TickIndex = self.orders.stored_tick(key) as u16;
            per_level
                .entry((idx, order.info.side, tick))
                .or_default()
                .push((order.info.created_at, oid, key, remaining));
        }

        for ((idx, _side, _tick), mut items) in per_level {
            items.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1)));
            for (_created_at, _oid, key, remaining) in items {
                self.runner_books[idx].insert_tail(&mut self.orders, key, remaining);
            }
        }
    }

    // ========================================================================
    // Accessors
    // ========================================================================

    pub fn market_id(&self) -> MarketId {
        self.market_id
    }

    pub fn market_kind(&self) -> MarketKind {
        self.market_kind
    }

    pub fn market_state(&self) -> BookMarketState {
        self.state
    }

    pub fn is_halted(&self) -> bool {
        self.state.is_halted()
    }

    pub fn get_order(&self, order_id: OrderId) -> Option<&BookOrder> {
        self.orders.get(&order_id)
    }

    pub fn get_trade(&self, trade_id: TradeId) -> Option<&BookTrade> {
        self.trades.get(&trade_id)
    }

    pub fn is_resting(&self, order_id: OrderId) -> bool {
        self.orders.is_in_level(&order_id)
    }

    pub fn active_order_count(&self) -> usize {
        self.orders
            .iter_sorted()
            .filter(|(_, o)| {
                matches!(
                    o.info.state,
                    BookOrderState::ExecutableUnmatched
                        | BookOrderState::ExecutablePartiallyMatched
                )
            })
            .count()
    }

    pub fn close_process_state(&self) -> Option<CloseProcessState> {
        self.close_process
    }

    pub fn runners(&self) -> impl Iterator<Item = RunnerId> + '_ {
        self.runner_ids.iter().copied()
    }

    // ========================================================================
    // Market Depth Queries
    // ========================================================================

    pub fn runner_prices(&self, runner_id: RunnerId, depth: usize) -> RunnerPrices {
        let mut result = RunnerPrices {
            runner_id,
            available_to_back: Vec::with_capacity(depth),
            available_to_lay: Vec::with_capacity(depth),
        };

        if depth == 0 {
            return result;
        }

        let Some(idx) = self.runner_index(runner_id) else {
            return result;
        };
        let opposite_idx = Self::opposite_index(idx);
        let runner_book = &self.runner_books[idx];
        let opposite_book = &self.runner_books[opposite_idx];

        // Available to back = direct LAY on this runner + implied BACK on opposite runner.
        // Both streams are iterated in ascending effective price order so we can merge and stop at depth.
        let mut direct_back_tick = runner_book.best_tick_asc(Side::No);
        let mut implied_back_tick = opposite_book.best_tick_desc(Side::Yes);
        let mut direct_back: Option<(OddsX10000, Money)> = None;
        let mut implied_back: Option<(OddsX10000, Money)> = None;

        while result.available_to_back.len() < depth {
            if direct_back.is_none() {
                while let Some(t) = direct_back_tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    let size = runner_book.level_total_remaining(Side::No, t);
                    direct_back_tick = runner_book.next_tick_asc_from(Side::No, t + 1);
                    if size.0 > 0 {
                        direct_back = Some((px, size));
                        break;
                    }
                }
            }

            if implied_back.is_none() {
                while let Some(t) = implied_back_tick {
                    let maker_px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    let size = opposite_book.level_total_remaining(Side::Yes, t);
                    implied_back_tick = if t == 0 {
                        None
                    } else {
                        opposite_book.next_tick_desc_from(Side::Yes, t - 1)
                    };
                    if size.0 > 0 {
                        implied_back = Some((Self::derived_odds(maker_px), size));
                        break;
                    }
                }
            }

            let next = match (direct_back, implied_back) {
                (Some(d), Some(i)) => {
                    if d.0.0 <= i.0.0 {
                        direct_back = None;
                        d
                    } else {
                        implied_back = None;
                        i
                    }
                }
                (Some(d), None) => {
                    direct_back = None;
                    d
                }
                (None, Some(i)) => {
                    implied_back = None;
                    i
                }
                (None, None) => break,
            };

            if let Some(last) = result.available_to_back.last_mut()
                && last.price == next.0
            {
                last.size = Money(last.size.0 + next.1.0);
                continue;
            }
            result.available_to_back.push(PriceSize {
                price: next.0,
                size: next.1,
            });
        }

        // Available to lay = direct BACK on this runner + implied LAY on opposite runner.
        // Both streams are iterated in descending effective price order so we can merge and stop at depth.
        let mut direct_lay_tick = runner_book.best_tick_desc(Side::Yes);
        let mut implied_lay_tick = opposite_book.best_tick_asc(Side::No);
        let mut direct_lay: Option<(OddsX10000, Money)> = None;
        let mut implied_lay: Option<(OddsX10000, Money)> = None;

        while result.available_to_lay.len() < depth {
            if direct_lay.is_none() {
                while let Some(t) = direct_lay_tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    let size = runner_book.level_total_remaining(Side::Yes, t);
                    direct_lay_tick = if t == 0 {
                        None
                    } else {
                        runner_book.next_tick_desc_from(Side::Yes, t - 1)
                    };
                    if size.0 > 0 {
                        direct_lay = Some((px, size));
                        break;
                    }
                }
            }

            if implied_lay.is_none() {
                while let Some(t) = implied_lay_tick {
                    let maker_px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    let size = opposite_book.level_total_remaining(Side::No, t);
                    implied_lay_tick = opposite_book.next_tick_asc_from(Side::No, t + 1);
                    if size.0 > 0 {
                        implied_lay = Some((Self::derived_odds(maker_px), size));
                        break;
                    }
                }
            }

            let next = match (direct_lay, implied_lay) {
                (Some(d), Some(i)) => {
                    if d.0.0 >= i.0.0 {
                        direct_lay = None;
                        d
                    } else {
                        implied_lay = None;
                        i
                    }
                }
                (Some(d), None) => {
                    direct_lay = None;
                    d
                }
                (None, Some(i)) => {
                    implied_lay = None;
                    i
                }
                (None, None) => break,
            };

            if let Some(last) = result.available_to_lay.last_mut()
                && last.price == next.0
            {
                last.size = Money(last.size.0 + next.1.0);
                continue;
            }
            result.available_to_lay.push(PriceSize {
                price: next.0,
                size: next.1,
            });
        }

        result
    }

    pub fn best_back_price(&self, runner_id: RunnerId) -> Option<PriceSize> {
        let prices = self.runner_prices(runner_id, 1);
        prices.available_to_back.into_iter().next()
    }

    pub fn best_lay_price(&self, runner_id: RunnerId) -> Option<PriceSize> {
        let prices = self.runner_prices(runner_id, 1);
        prices.available_to_lay.into_iter().next()
    }

    pub fn runner_matched_volume(&self, runner_id: RunnerId) -> Money {
        self.runner_index(runner_id)
            .map(|idx| self.runner_matched_volume[idx])
            .unwrap_or(Money::zero())
    }

    pub fn total_matched(&self) -> Money {
        Money(self.runner_matched_volume[0].0 + self.runner_matched_volume[1].0)
    }

    // ========================================================================
    // Command Processing
    // ========================================================================

    pub(crate) fn handle_command(
        &mut self,
        cmd: &Command,
    ) -> Result<(EventVec, CommandResponse), BookError> {
        let correlation_id = cmd.correlation_id;
        let err = |reason| BookError::new(correlation_id, reason);
        let ts = Utc::now();
        let market_id = cmd.market_id;
        if market_id != self.market_id {
            return Err(err(RejectReason::MarketIdMismatch));
        }

        match &cmd.kind {
            CommandKind::HaltMarket { reason, .. } => {
                let events = self.cmd_halt_market(ts, *reason).map_err(&err)?;
                return Ok((
                    events,
                    CommandResponse {
                        correlation_id,
                        kind: None,
                    },
                ));
            }
            CommandKind::ResumeMarket => {
                let events = self.cmd_resume_market(ts).map_err(&err)?;
                return Ok((
                    events,
                    CommandResponse {
                        correlation_id,
                        kind: None,
                    },
                ));
            }
            _ => {}
        }

        if self.state.is_halted() {
            return Err(err(RejectReason::MarketHalted));
        }

        let mut scratch = std::mem::take(&mut self.scratch);
        let result = (|| -> Result<(EventVec, CommandResponse), BookError> {
            let (events, kind) = match &cmd.kind {
                CommandKind::CreateMarket { .. } => {
                    return Err(err(RejectReason::InternalError));
                }
                CommandKind::SetMarketState { state, .. } => {
                    let reason = "SET_MARKET_STATE";
                    match state {
                        MarketState::Open => {
                            if self.state == BookMarketState::Suspended {
                                self.cmd_unsuspend(&mut scratch, ts, reason).map_err(&err)?
                            } else {
                                self.cmd_open_market(&mut scratch, ts, reason)
                                    .map_err(&err)?
                            }
                        }
                        MarketState::InPlay => self
                            .cmd_turn_in_play(&mut scratch, ts, reason)
                            .map_err(&err)?,
                        MarketState::Suspended => {
                            self.cmd_suspend(&mut scratch, ts, reason).map_err(&err)?
                        }
                        MarketState::Closed => self
                            .cmd_close_market(ts, crate::config::DEFAULT_SET_MARKET_STATE_BATCH)
                            .map_err(&err)?,
                    }
                }

                CommandKind::CloseMarket {
                    batch_max_events, ..
                } => self.cmd_close_market(ts, *batch_max_events).map_err(&err)?,

                CommandKind::ContinueCloseMarket => self
                    .cmd_continue_close_market(&mut scratch, ts)
                    .map_err(&err)?,

                CommandKind::VoidMarket { reason, .. } => self
                    .cmd_void_market(&mut scratch, ts, reason.as_str())
                    .map_err(&err)?,

                CommandKind::SettleMarket {
                    runner_results,
                    dead_heat_divisor,
                    ..
                } => self
                    .cmd_settle_market(
                        &mut scratch,
                        ts,
                        runner_results.as_slice(),
                        *dead_heat_divisor,
                    )
                    .map_err(&err)?,

                CommandKind::PlaceOrder {
                    account_id,
                    runner_id,
                    side,
                    odds,
                    stake,
                    persistence,
                    time_in_force,
                    ..
                } => {
                    let order_id = OrderId(self.next_order_id);
                    self.cmd_place_order(
                        &mut scratch,
                        ts,
                        order_id,
                        *account_id,
                        *runner_id,
                        *side,
                        *odds,
                        *stake,
                        *persistence,
                        *time_in_force,
                    )
                    .map_err(&err)?
                }

                CommandKind::PlaceBinaryOrder { .. } | CommandKind::ReplaceBinaryOrder { .. } => {
                    return Err(err(RejectReason::MarketModelMismatch));
                }

                CommandKind::CancelOrder {
                    order_id,
                    account_id,
                    ..
                } => self
                    .cmd_cancel_order(&mut scratch, ts, *order_id, *account_id, "USER_CANCEL")
                    .map_err(&err)?,

                CommandKind::ReplaceOrder {
                    order_id,
                    account_id,
                    new_odds,
                    new_stake,
                    ..
                } => {
                    let Some(old) = self.get_order(*order_id) else {
                        return Err(err(RejectReason::OrderNotFound));
                    };
                    if old.info.account_id != *account_id {
                        return Err(err(RejectReason::NotOrderOwner));
                    }
                    let new_price = new_odds.unwrap_or(old.price);
                    let new_stake = new_stake.unwrap_or_else(|| old.remaining());
                    let new_order_id = OrderId(self.next_order_id);
                    self.cmd_replace_order(
                        &mut scratch,
                        ts,
                        *order_id,
                        new_order_id,
                        *account_id,
                        old.runner_id,
                        old.info.side,
                        new_price,
                        new_stake,
                        old.persistence,
                    )
                    .map_err(&err)?
                }
                CommandKind::RemoveRunner {
                    runner_id,
                    reduction_factor_bps,
                    ..
                } => self
                    .cmd_remove_runner(&mut scratch, ts, *runner_id, *reduction_factor_bps)
                    .map_err(&err)?,
                CommandKind::CashoutRunner { .. } => {
                    return Err(err(RejectReason::CashoutNotSupported));
                }

                CommandKind::VoidTradesFromTime {
                    from_matched_at_inclusive,
                    reason,
                    ..
                } => self
                    .cmd_void_trades_from_time(
                        &mut scratch,
                        ts,
                        *from_matched_at_inclusive,
                        reason.as_str(),
                    )
                    .map_err(&err)?,

                CommandKind::VoidTradeIds {
                    trade_ids, reason, ..
                } => self
                    .cmd_void_trade_ids(&mut scratch, ts, trade_ids.as_slice(), reason.as_str())
                    .map_err(&err)?,

                CommandKind::HaltMarket { .. } | CommandKind::ResumeMarket => {
                    unreachable!("handled above")
                }
                CommandKind::RemoveMarket => {
                    return Err(err(RejectReason::InternalError));
                }
            };

            Ok((
                events,
                CommandResponse {
                    correlation_id,
                    kind,
                },
            ))
        })();

        self.scratch = scratch;
        result
    }

    // ========================================================================
    // Internal Helpers
    // ========================================================================

    pub(crate) fn apply_event(&mut self, env: &BookEventEnvelope) {
        let ts = env.timestamp;
        let event = &env.event;
        match event {
            BookEvent::MarketCreated { .. } => {}
            BookEvent::MarketStateChanged { to, reason } => {
                self.market_state_changed(ts, *to, reason.as_str())
            }
            BookEvent::OrderAccepted {
                order_id,
                account_id,
                runner_id,
                side,
                price,
                stake,
                persistence,
                time_in_force,
            } => {
                self.next_order_id = self.next_order_id.max(order_id.0.saturating_add(1));
                self.order_accepted(
                    ts,
                    *order_id,
                    *account_id,
                    *runner_id,
                    *side,
                    *price,
                    *stake,
                    *persistence,
                    *time_in_force,
                )
            }
            BookEvent::BinaryOrderAccepted { .. } => {}
            BookEvent::OrderCancelled {
                order_id, reason, ..
            } => self.order_cancelled(ts, *order_id, reason.as_str()),
            BookEvent::OrderLapsed {
                order_id,
                reason: _,
            } => {
                self.order_lapsed(ts, *order_id);
            }
            BookEvent::OrderVoided {
                order_id,
                reason: _,
            } => {
                self.order_voided(ts, *order_id);
            }
            BookEvent::TradeMatched {
                trade_id,
                maker,
                taker,
                runner_id,
                price,
                stake,
            } => self.trade_matched(TradeMatch {
                ts,
                trade_id: *trade_id,
                maker_order_id: maker.id,
                taker_order_id: taker.id,
                taker_runner_id: *runner_id,
                price: *price,
                stake: *stake,
            }),
            BookEvent::BinaryTradeMatched { .. } => {}
            BookEvent::TradeVoided { trade_id, reason } => {
                self.trade_voided(ts, *trade_id, reason.as_str())
            }
            BookEvent::RunnerRemoved { runner_id, .. } => {
                if self.runner_index(*runner_id).is_some() {
                    self.removed_runners.insert(*runner_id);
                }
            }
            BookEvent::MarketSettled { .. } => {
                // No book state beyond the terminal MarketStateChanged; downstream uses results.
            }
            BookEvent::MarketOrdersSettled {
                cursor_after,
                order_ids,
                is_final,
            } => {
                for &order_id in order_ids {
                    self.order_cancelled(ts, order_id, "CLOSE_PROCESS");
                }
                if let Some(s) = self.close_process.as_mut() {
                    s.cursor_after = *cursor_after;
                    s.cancelled_total = s.cancelled_total.saturating_add(order_ids.len() as u64);
                    s.chunks_done = s.chunks_done.saturating_add(1);
                    if *is_final {
                        // Retain final close stats so consumers can observe completion (e.g. UI).
                    }
                }
            }

            BookEvent::OrderRejected { .. } => {}
            BookEvent::MarketRemoved { .. } => {}
        }
    }

    pub(crate) fn market_state_changed(
        &mut self,
        _ts: DateTime,
        to: BookMarketState,
        reason: &str,
    ) {
        if self.market_kind == MarketKind::PreEventOnly {
            assert!(
                to != BookMarketState::TurnInPlayEnabled,
                "pre-event-only market cannot be applied into TurnInPlayEnabled"
            );
        }
        if to == BookMarketState::Closing {
            let batch_max_events =
                crate::book::close_process::parse_close_start_batch_max_events(reason)
                    .unwrap_or(crate::book::close_process::DEFAULT_CLOSE_BATCH_EVENTS);
            let total_live_orders =
                crate::book::close_process::count_live_orders(self.orders.iter_sorted());
            self.close_process = Some(CloseProcessState {
                batch_max_events,
                cursor_after: None,
                total_live_orders,
                cancelled_total: 0,
                chunks_done: 0,
            });
        }
        let from = self.state;
        if to == BookMarketState::Suspended {
            self.state_before_suspend = Some(from);
        } else if from == BookMarketState::Suspended {
            self.state_before_suspend = None;
        }
        if to == BookMarketState::Halted {
            self.state_before_halt = Some(from);
        } else if from == BookMarketState::Halted {
            self.state_before_halt = None;
        }
        self.state = to;
    }

    #[allow(clippy::too_many_arguments)]
    pub(crate) fn order_accepted(
        &mut self,
        ts: DateTime,
        order_id: OrderId,
        account_id: AccountId,
        runner_id: RunnerId,
        side: Side,
        price: OddsX10000,
        stake: Money,
        persistence: Persistence,
        _time_in_force: TimeInForce,
    ) {
        if self.orders.contains(&order_id) {
            return;
        }
        let Some(runner_idx) = self.runner_index(runner_id) else {
            return;
        };

        if let Err(err) = self.orders.insert(
            order_id,
            BookOrder {
                info: BookOrderInfo {
                    order_id,
                    account_id,
                    side,
                    state: BookOrderState::ExecutableUnmatched,
                    created_at: ts,
                    last_updated_at: ts,
                },
                runner_id,
                price,
                stake,
                matched: Money::zero(),
                persistence,
            },
        ) {
            error!(order_id = ?order_id, reason = ?err, "failed to insert order");
            return;
        }

        let Some(key) = self.orders.get_key(&order_id) else {
            error!(order_id = ?order_id, "missing order key after insert");
            return;
        };
        self.runner_books[runner_idx].insert_tail(&mut self.orders, key, stake);
    }

    fn trade_matched(&mut self, m: TradeMatch) {
        // Update maker and taker order quantities/states.
        let maker_runner_id = self
            .orders
            .get(&m.maker_order_id)
            .unwrap_or_else(|| {
                panic!(
                    "trade_matched: maker order {:?} not found for trade {:?}",
                    m.maker_order_id, m.trade_id
                )
            })
            .runner_id;
        let maker_runner_idx = self.runner_index(maker_runner_id).unwrap_or_else(|| {
            panic!(
                "trade_matched: maker runner {:?} not found for trade {:?}",
                maker_runner_id, m.trade_id
            )
        });
        let taker_runner_idx = self.runner_index(m.taker_runner_id).unwrap_or_else(|| {
            panic!(
                "trade_matched: taker runner {:?} not found for trade {:?}",
                m.taker_runner_id, m.trade_id
            )
        });

        let is_implied = maker_runner_id != m.taker_runner_id;
        if let Some(maker) = self.orders.get_mut(&m.maker_order_id) {
            maker.matched = Money(maker.matched.0 + m.stake.0);
            maker.info.last_updated_at = m.ts;
            maker.info.state = if maker.remaining().0 == 0 {
                BookOrderState::ExecutionComplete
            } else {
                BookOrderState::ExecutablePartiallyMatched
            };
        }

        if let Some(taker) = self.orders.get_mut(&m.taker_order_id) {
            taker.matched = Money(taker.matched.0 + m.stake.0);
            taker.info.last_updated_at = m.ts;
            taker.info.state = if taker.remaining().0 == 0 {
                BookOrderState::ExecutionComplete
            } else {
                BookOrderState::ExecutablePartiallyMatched
            };
        }

        // Keep ladder totals in sync for whichever side was resting at the moment.
        self.update_level_after_fill(m.maker_order_id, maker_runner_idx, m.stake, is_implied);
        self.update_level_after_fill(m.taker_order_id, taker_runner_idx, m.stake, false);

        // Terminal orders are not retained in the book.
        for oid in [m.maker_order_id, m.taker_order_id] {
            let remaining = self
                .orders
                .get(&oid)
                .map(|o| o.remaining())
                .unwrap_or(Money::zero());
            if remaining.0 == 0 {
                let _ = self.orders.remove(&oid);
            }
        }

        self.runner_matched_volume[taker_runner_idx] =
            Money(self.runner_matched_volume[taker_runner_idx].0 + m.stake.0);
        if is_implied {
            self.runner_matched_volume[maker_runner_idx] =
                Money(self.runner_matched_volume[maker_runner_idx].0 + m.stake.0);
        }

        self.next_trade_id = self.next_trade_id.max(m.trade_id.0.saturating_add(1));
        self.trades.entry(m.trade_id).or_insert(BookTrade {
            trade_id: m.trade_id,
            maker_order_id: m.maker_order_id,
            taker_order_id: m.taker_order_id,
            runner_id: m.taker_runner_id,
            price: m.price,
            stake: m.stake,
            matched_at: m.ts,
            state: BookTradeState::Live,
        });
    }

    pub(crate) fn trade_voided(&mut self, ts: DateTime, trade_id: TradeId, _reason: &str) {
        let t = self
            .trades
            .get_mut(&trade_id)
            .unwrap_or_else(|| panic!("trade_voided: trade {:?} not found", trade_id));
        if t.state == BookTradeState::Voided {
            return;
        }

        let stake = t.stake;
        let taker_runner_id = t.runner_id;
        let maker_runner_id = self.orders.get(&t.maker_order_id).map(|o| o.runner_id);
        t.state = BookTradeState::Voided;

        // Betfair-style: keep the remaining (unmatched) portion unchanged, void only the matched
        // portion. Implemented by reducing both `stake` and `matched` by the same amount.
        for oid in [t.maker_order_id, t.taker_order_id] {
            let Some(o) = self.orders.get_mut(&oid) else {
                continue;
            };
            o.matched = Money(o.matched.0.saturating_sub(stake.0));
            o.stake = Money(o.stake.0.saturating_sub(stake.0));
            o.info.last_updated_at = ts;

            match o.info.state {
                BookOrderState::ExecutableUnmatched
                | BookOrderState::ExecutablePartiallyMatched
                | BookOrderState::ExecutionComplete => {
                    let remaining = o.remaining();
                    o.info.state = if remaining.0 == 0 {
                        BookOrderState::ExecutionComplete
                    } else if o.matched.0 == 0 {
                        BookOrderState::ExecutableUnmatched
                    } else {
                        BookOrderState::ExecutablePartiallyMatched
                    };
                }
                _ => {}
            }
        }

        if let Some(taker_idx) = self.runner_index(taker_runner_id) {
            self.runner_matched_volume[taker_idx] = Money(
                self.runner_matched_volume[taker_idx]
                    .0
                    .saturating_sub(stake.0),
            );
        }
        if let Some(maker_runner_id) = maker_runner_id
            && maker_runner_id != taker_runner_id
            && let Some(maker_idx) = self.runner_index(maker_runner_id)
        {
            self.runner_matched_volume[maker_idx] = Money(
                self.runner_matched_volume[maker_idx]
                    .0
                    .saturating_sub(stake.0),
            );
        }
    }

    pub(crate) fn order_cancelled(&mut self, ts: DateTime, order_id: OrderId, _reason: &str) {
        self.remove_from_levels(order_id);
        let _ = ts;
        let _ = self.orders.remove(&order_id);
    }

    pub(crate) fn order_lapsed(&mut self, ts: DateTime, order_id: OrderId) {
        self.remove_from_levels(order_id);
        let _ = ts;
        let _ = self.orders.remove(&order_id);
    }

    pub(crate) fn order_voided(&mut self, ts: DateTime, order_id: OrderId) {
        self.remove_from_levels(order_id);
        let _ = ts;
        let _ = self.orders.remove(&order_id);
    }

    fn emit(&self, event_time: DateTime, event: BookEvent) -> BookEventEnvelope {
        BookEventEnvelope {
            market_id: self.market_id,
            market_seq: 0,
            timestamp: event_time,
            event,
        }
    }

    fn state_change(&self, ts: DateTime, to: BookMarketState, reason: &str, out: &mut EventVec) {
        if self.state == to {
            return;
        }
        if self.market_kind == MarketKind::PreEventOnly {
            assert!(
                to != BookMarketState::TurnInPlayEnabled,
                "pre-event-only market cannot transition to TurnInPlayEnabled"
            );
        }
        out.push(self.emit(
            ts,
            BookEvent::MarketStateChanged {
                to,
                reason: reason.to_string(),
            },
        ));
    }

    fn ok_noop(events: EventVec) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        Ok((events, None))
    }

    // ========================================================================
    // Market State Commands
    // ========================================================================

    fn cmd_open_market(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        reason: &'static str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();
        if self.state.is_terminal() {
            return Err(RejectReason::MarketTerminal);
        }
        self.state_change(ts, BookMarketState::Open, reason, &mut scratch.events);
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    fn cmd_turn_in_play(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        reason: &'static str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();
        if self.market_kind == MarketKind::PreEventOnly {
            return Err(RejectReason::MarketInPlayNotSupported);
        }
        if self.state.is_terminal() {
            return Err(RejectReason::MarketTerminal);
        }

        if self.state == BookMarketState::Open {
            let lapse_ids: Vec<OrderId> = self
                .orders
                .iter_sorted()
                .filter_map(|(oid, o)| {
                    let is_live = matches!(
                        o.info.state,
                        BookOrderState::ExecutableUnmatched
                            | BookOrderState::ExecutablePartiallyMatched
                    );
                    if is_live && o.persistence == Persistence::Lapse {
                        Some(oid)
                    } else {
                        None
                    }
                })
                .collect();

            for oid in lapse_ids {
                scratch.events.push(self.emit(
                    ts,
                    BookEvent::OrderLapsed {
                        order_id: oid,
                        reason: "IN_PLAY_LAPSE".to_string(),
                    },
                ));
            }
        }

        self.state_change(
            ts,
            BookMarketState::TurnInPlayEnabled,
            reason,
            &mut scratch.events,
        );
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    fn cmd_suspend(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        reason: &'static str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();
        if !self.state.is_matchable() {
            return Err(RejectReason::MarketNotOpen);
        }

        self.state_change(ts, BookMarketState::Suspended, reason, &mut scratch.events);
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    fn cmd_unsuspend(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        reason: &'static str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();
        if self.state != BookMarketState::Suspended {
            return Err(RejectReason::MarketNotSuspended);
        }
        let restored = self.state_before_suspend.unwrap_or(BookMarketState::Open);
        self.state_change(ts, restored, reason, &mut scratch.events);
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    fn cmd_halt_market(&self, ts: DateTime, reason: u32) -> Result<EventVec, RejectReason> {
        if self.state.is_terminal() {
            return Err(RejectReason::MarketTerminal);
        }
        if self.state.is_halted() {
            return Err(RejectReason::MarketAlreadyHalted);
        }
        let mut events = EventVec::new();
        self.state_change(
            ts,
            BookMarketState::Halted,
            &format!("HALT:{reason}"),
            &mut events,
        );
        Ok(events)
    }

    fn cmd_resume_market(&self, ts: DateTime) -> Result<EventVec, RejectReason> {
        if !self.state.is_halted() {
            return Err(RejectReason::MarketNotHalted);
        }
        let restored = self.state_before_halt.unwrap_or(BookMarketState::Open);
        let mut events = EventVec::new();
        self.state_change(ts, restored, "RESUME", &mut events);
        Ok(events)
    }

    fn cmd_close_market(
        &mut self,
        ts: DateTime,
        batch_max_events: u16,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        // `close_process` may be retained after completion for UI/debugging; state is authoritative.
        if self.state == BookMarketState::Closing {
            return Err(RejectReason::MarketClosing);
        }
        if self.state.is_terminal() {
            return Err(RejectReason::MarketTerminal);
        }
        if batch_max_events < MIN_CLOSE_BATCH_EVENTS {
            return Err(RejectReason::InvalidBatchSize);
        }

        let events = close_start_batch(
            batch_max_events,
            |to, reason, out: &mut Vec<BookEventEnvelope>| {
                let mut ev: EventVec = EventVec::from_vec(std::mem::take(out));
                self.state_change(ts, to, reason, &mut ev);
                *out = ev.into_vec();
            },
            |e| self.emit(ts, e),
            |cursor_after, max_orders| {
                let mut cancelled_order_ids = Vec::new();
                let mut last = cursor_after;
                let mut cancelled = 0usize;

                let mut it = self.orders.iter_sorted_from(cursor_after).peekable();
                while let Some((oid, order)) = it.next() {
                    last = Some(oid);
                    if !crate::book::close_process::is_live_order(order) {
                        continue;
                    }
                    cancelled_order_ids.push(oid);
                    cancelled += 1;
                    if cancelled >= max_orders {
                        let done = it.peek().is_none();
                        return (cancelled_order_ids, last, done, cancelled as u64);
                    }
                }

                (cancelled_order_ids, last, true, cancelled as u64)
            },
        );

        // debug_assert!(events.len() <= batch_max_events as usize);
        Self::ok_noop(EventVec::from_vec(events))
    }

    fn cmd_continue_close_market(
        &self,
        _scratch: &mut Scratch,
        ts: DateTime,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        if self.state != BookMarketState::Closing {
            return Err(RejectReason::MarketNotClosing);
        }
        let Some(proc_state) = self.close_process else {
            return Err(RejectReason::InternalError);
        };

        let events = close_continue_batch(
            proc_state,
            |to, reason, out: &mut Vec<BookEventEnvelope>| {
                let mut ev: EventVec = EventVec::from_vec(std::mem::take(out));
                self.state_change(ts, to, reason, &mut ev);
                *out = ev.into_vec();
            },
            |e| self.emit(ts, e),
            |cursor_after, max_orders| {
                let mut cancelled_order_ids = Vec::new();
                let mut last = cursor_after;
                let mut cancelled = 0usize;

                let mut it = self.orders.iter_sorted_from(cursor_after).peekable();
                while let Some((oid, order)) = it.next() {
                    last = Some(oid);
                    if !crate::book::close_process::is_live_order(order) {
                        continue;
                    }
                    cancelled_order_ids.push(oid);
                    cancelled += 1;
                    if cancelled >= max_orders {
                        let done = it.peek().is_none();
                        return (cancelled_order_ids, last, done, cancelled as u64);
                    }
                }

                (cancelled_order_ids, last, true, cancelled as u64)
            },
        );

        debug_assert!(events.len() <= proc_state.batch_max_events as usize);
        Self::ok_noop(EventVec::from_vec(events))
    }

    fn cmd_void_market(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        reason: &str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();
        if self.state == BookMarketState::Voided {
            return Err(RejectReason::MarketAlreadyVoided);
        }

        let live_orders: Vec<OrderId> = self
            .orders
            .iter_sorted()
            .filter_map(|(oid, o)| {
                if matches!(
                    o.info.state,
                    BookOrderState::ExecutableUnmatched
                        | BookOrderState::ExecutablePartiallyMatched
                ) {
                    Some(oid)
                } else {
                    None
                }
            })
            .collect();

        for oid in live_orders {
            scratch.events.push(self.emit(
                ts,
                BookEvent::OrderVoided {
                    order_id: oid,
                    reason: "MARKET_VOID".to_string(),
                },
            ));
        }

        let trade_ids: Vec<TradeId> = self
            .trades
            .iter()
            .filter_map(|(&tid, t)| (t.state == BookTradeState::Live).then_some(tid))
            .collect();

        for tid in trade_ids {
            scratch.events.push(self.emit(
                ts,
                BookEvent::TradeVoided {
                    trade_id: tid,
                    reason: "MARKET_VOID".to_string(),
                },
            ));
        }

        self.state_change(ts, BookMarketState::Voided, reason, &mut scratch.events);
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    // ========================================================================
    // Order Commands
    // ========================================================================

    #[allow(clippy::too_many_arguments)]
    fn cmd_place_order(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        order_id: OrderId,
        account_id: AccountId,
        runner_id: RunnerId,
        side: Side,
        price: OddsX10000,
        stake: Money,
        persistence: Persistence,
        time_in_force: TimeInForce,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        if !self.state.is_matchable() {
            return Err(RejectReason::MarketNotOpen);
        }

        let Some(runner_idx) = self.runner_index(runner_id) else {
            return Err(RejectReason::RunnerNotFound);
        };

        if self.removed_runners.contains(&runner_id) {
            return Err(RejectReason::RunnerNotFound);
        }

        if !price.is_valid_tick() {
            return Err(RejectReason::InvalidOdds);
        }

        if !stake.is_positive() {
            return Err(RejectReason::InvalidStake);
        }

        if self.orders.contains(&order_id) {
            return Err(RejectReason::Duplicate);
        }

        if let TimeInForce::FillOrKill { min_fill } = time_in_force {
            let required = match min_fill {
                Some(qty) => {
                    let value = i64::try_from(qty.0).map_err(|_| RejectReason::InvalidStake)?;
                    Money(value)
                }
                None => stake,
            };
            let available = self.available_to_match(runner_idx, side, price);
            if available.0 < required.0 {
                return Err(RejectReason::WouldNotFillFok);
            }
        }

        scratch.events.push(self.emit(
            ts,
            BookEvent::OrderAccepted {
                order_id,
                account_id,
                runner_id,
                side,
                price,
                stake,
                persistence,
                time_in_force,
            },
        ));

        let (matched, avg_price) = self.plan_matching(
            scratch,
            MatchRequest {
                ts,
                taker_order_id: order_id,
                taker_account_id: account_id,
                taker_side: side,
                taker_runner_id: runner_id,
                taker_price: price,
                taker_stake: stake,
                taker_runner_idx: runner_idx,
            },
        );

        let mut final_remaining = Money(stake.0.saturating_sub(matched.0));
        let mut final_state = if final_remaining.0 == 0 {
            BookOrderState::ExecutionComplete
        } else if matched.0 > 0 {
            BookOrderState::ExecutablePartiallyMatched
        } else {
            BookOrderState::ExecutableUnmatched
        };

        let should_cancel_remainder =
            matches!(time_in_force, TimeInForce::FillOrKill { .. }) && final_remaining.0 > 0;
        if should_cancel_remainder {
            final_remaining = Money::zero();
            final_state = BookOrderState::Cancelled;
            scratch.events.push(self.emit(
                ts,
                BookEvent::OrderCancelled {
                    order_id,
                    reason: "FOK_REMAINDER".to_string(),
                },
            ));
        }

        Ok((
            std::mem::take(&mut scratch.events),
            Some(CommandResponseKind::PlaceOrder(PlaceOrderResult {
                accepted: true,
                order_id,
                matched: FillQuantity::Stake(matched),
                avg_matched_price: avg_price.map(FillPrice::Odds),
                remaining: FillQuantity::Stake(final_remaining),
                final_order_state: Some(final_state),
            })),
        ))
    }

    /// Calculate total available liquidity to match against for a given taker order.
    fn available_to_match(&self, runner_idx: usize, side: Side, price: OddsX10000) -> Money {
        let opposite_idx = Self::opposite_index(runner_idx);
        let derived_price = Self::derived_odds(price);
        let mut total = 0i64;

        match side {
            Side::Yes => {
                // Direct: LAY orders on same runner at price <= taker_price (ascending)
                let rb = &self.runner_books[runner_idx];
                let mut tick = rb.best_tick_asc(Side::No);
                while let Some(t) = tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if px.0 > price.0 {
                        break;
                    }
                    total += rb.level_total_remaining(Side::No, t).0;
                    tick = rb.next_tick_asc_from(Side::No, t + 1);
                }

                // Implied: BACK on opposite runner where maker_price <= derived(taker_price)
                let rb = &self.runner_books[opposite_idx];
                let mut tick = rb.best_tick_asc(Side::Yes);
                while let Some(t) = tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if px.0 > derived_price.0 {
                        break;
                    }
                    total += rb.level_total_remaining(Side::Yes, t).0;
                    tick = rb.next_tick_asc_from(Side::Yes, t + 1);
                }
            }
            Side::No => {
                // Direct: BACK orders on same runner at price >= taker_price (descending)
                let rb = &self.runner_books[runner_idx];
                let mut tick = rb.best_tick_desc(Side::Yes);
                while let Some(t) = tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if px.0 < price.0 {
                        break;
                    }
                    total += rb.level_total_remaining(Side::Yes, t).0;
                    if t == 0 {
                        break;
                    }
                    tick = rb.next_tick_desc_from(Side::Yes, t - 1);
                }

                // Implied: LAY on opposite runner where maker_price >= derived(taker_price)
                let rb = &self.runner_books[opposite_idx];
                let mut tick = rb.best_tick_desc(Side::No);
                while let Some(t) = tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if px.0 < derived_price.0 {
                        break;
                    }
                    total += rb.level_total_remaining(Side::No, t).0;
                    if t == 0 {
                        break;
                    }
                    tick = rb.next_tick_desc_from(Side::No, t - 1);
                }
            }
        }

        Money(total)
    }

    /// Plan matching with implied liquidity without mutating book state.
    fn plan_matching(&self, scratch: &mut Scratch, r: MatchRequest) -> (Money, Option<OddsX10000>) {
        let opposite_idx = Self::opposite_index(r.taker_runner_idx);
        let derived_taker_price = Self::derived_odds(r.taker_price);

        scratch.maker_matched_delta.clear();
        scratch.matchable.clear();
        scratch.fills.clear();

        match r.taker_side {
            Side::Yes => {
                // Direct: LAY on same runner (ascending)
                let rb = &self.runner_books[r.taker_runner_idx];
                let mut tick = rb.best_tick_asc(Side::No);
                while let Some(t) = tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if px.0 > r.taker_price.0 {
                        break;
                    }
                    for maker_key in rb.iter_level_keys(Side::No, t, &self.orders) {
                        let maker = self.orders.order_by_key(maker_key);
                        scratch.matchable.push(MatchableOrder {
                            order_id: maker.info.order_id,
                            effective_price: px,
                            created_at: maker.info.created_at,
                        });
                    }
                    tick = rb.next_tick_asc_from(Side::No, t + 1);
                }

                // Implied: BACK on opposite runner where maker_price <= derived(taker_price) (ascending)
                let rb = &self.runner_books[opposite_idx];
                let mut tick = rb.best_tick_asc(Side::Yes);
                while let Some(t) = tick {
                    let maker_px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if maker_px.0 > derived_taker_price.0 {
                        break;
                    }
                    let effective = Self::derived_odds(maker_px);
                    for maker_key in rb.iter_level_keys(Side::Yes, t, &self.orders) {
                        let maker = self.orders.order_by_key(maker_key);
                        scratch.matchable.push(MatchableOrder {
                            order_id: maker.info.order_id,
                            effective_price: effective,
                            created_at: maker.info.created_at,
                        });
                    }
                    tick = rb.next_tick_asc_from(Side::Yes, t + 1);
                }

                // Sort by effective price ascending (best price first), then by order creation time
                scratch.matchable.sort_by(|a, b| {
                    a.effective_price
                        .0
                        .cmp(&b.effective_price.0)
                        // Deterministic tie-breakers:
                        // - `created_at` for FIFO within the same effective price
                        // - `order_id` to break ties when timestamps collide
                        .then_with(|| a.created_at.cmp(&b.created_at))
                        .then_with(|| a.order_id.cmp(&b.order_id))
                });
            }
            Side::No => {
                // Direct: BACK on same runner (descending)
                let rb = &self.runner_books[r.taker_runner_idx];
                let mut tick = rb.best_tick_desc(Side::Yes);
                while let Some(t) = tick {
                    let px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if px.0 < r.taker_price.0 {
                        break;
                    }
                    for maker_key in rb.iter_level_keys(Side::Yes, t, &self.orders) {
                        let maker = self.orders.order_by_key(maker_key);
                        scratch.matchable.push(MatchableOrder {
                            order_id: maker.info.order_id,
                            effective_price: px,
                            created_at: maker.info.created_at,
                        });
                    }
                    if t == 0 {
                        break;
                    }
                    tick = rb.next_tick_desc_from(Side::Yes, t - 1);
                }

                // Implied: LAY on opposite runner where maker_price >= derived(taker_price) (descending)
                let rb = &self.runner_books[opposite_idx];
                let mut tick = rb.best_tick_desc(Side::No);
                while let Some(t) = tick {
                    let maker_px = OddsX10000(crate::types::odds::TICK_LADDER[t]);
                    if maker_px.0 < derived_taker_price.0 {
                        break;
                    }
                    let effective = Self::derived_odds(maker_px);
                    for maker_key in rb.iter_level_keys(Side::No, t, &self.orders) {
                        let maker = self.orders.order_by_key(maker_key);
                        scratch.matchable.push(MatchableOrder {
                            order_id: maker.info.order_id,
                            effective_price: effective,
                            created_at: maker.info.created_at,
                        });
                    }
                    if t == 0 {
                        break;
                    }
                    tick = rb.next_tick_desc_from(Side::No, t - 1);
                }

                // Sort by effective price descending (best price first), then by order creation time
                scratch.matchable.sort_by(|a, b| {
                    b.effective_price
                        .0
                        .cmp(&a.effective_price.0)
                        // Deterministic tie-breakers:
                        // - `created_at` for FIFO within the same effective price
                        // - `order_id` to break ties when timestamps collide
                        .then_with(|| a.created_at.cmp(&b.created_at))
                        .then_with(|| a.order_id.cmp(&b.order_id))
                });
            }
        }

        let mut remaining = r.taker_stake;
        let mut matched_total = Money::zero();
        let mut sum_price_qty: u128 = 0;

        for m in scratch.matchable.iter() {
            if remaining.0 <= 0 {
                break;
            }

            let Some(maker) = self.orders.get(&m.order_id) else {
                continue;
            };

            // Skip self-match
            if maker.info.account_id == r.taker_account_id {
                continue;
            }

            let maker_is_live = matches!(
                maker.info.state,
                BookOrderState::ExecutableUnmatched | BookOrderState::ExecutablePartiallyMatched
            );
            if !maker_is_live {
                continue;
            }

            let base_remaining = maker.stake.0.saturating_sub(maker.matched.0);
            let delta = *scratch.maker_matched_delta.get(&m.order_id).unwrap_or(&0);
            let maker_remaining = base_remaining.saturating_sub(delta);
            if maker_remaining <= 0 {
                continue;
            }

            let fill_i64 = remaining.0.min(maker_remaining);
            let fill_amount = Money(fill_i64);
            remaining = Money(remaining.0 - fill_i64);

            // Trade price: for direct matches, use effective price.
            // For implied matches, use the derived price of the maker's actual price.
            let trade_price = m.effective_price;

            *scratch.maker_matched_delta.entry(m.order_id).or_insert(0) += fill_i64;
            scratch.fills.push(PlannedFill {
                maker_order_id: m.order_id,
                maker_side: maker.info.side,
                trade_price,
                fill_amount,
            });

            matched_total = Money(matched_total.0 + fill_i64);
            sum_price_qty =
                sum_price_qty.saturating_add(trade_price.0 as u128 * fill_i64.max(0) as u128);
        }

        let mut next_trade_id = self.next_trade_id;
        for fill in scratch.fills.iter().copied() {
            let trade_id = TradeId(next_trade_id);
            next_trade_id = next_trade_id.saturating_add(1);
            scratch.events.push(self.emit(
                r.ts,
                BookEvent::TradeMatched {
                    trade_id,
                    maker: OrderInfo {
                        id: fill.maker_order_id,
                        side: fill.maker_side,
                    },
                    taker: OrderInfo {
                        id: r.taker_order_id,
                        side: r.taker_side,
                    },
                    runner_id: r.taker_runner_id,
                    price: fill.trade_price,
                    stake: fill.fill_amount,
                },
            ));
        }

        let avg_price = if matched_total.0 > 0 {
            let denom = matched_total.0 as u128;
            let px = (sum_price_qty / denom).min(u32::MAX as u128) as u32;
            Some(OddsX10000(px))
        } else {
            None
        };

        (matched_total, avg_price)
    }

    fn update_level_after_fill(
        &mut self,
        maker_order_id: OrderId,
        runner_idx: usize,
        fill_amount: Money,
        _is_implied: bool,
    ) {
        let (maker_side, maker_remaining) = match self.orders.get(&maker_order_id) {
            Some(o) => (o.info.side, o.remaining()),
            None => return,
        };
        let Some(maker_key) = self.orders.get_key(&maker_order_id) else {
            return;
        };
        let maker_tick = self.orders.stored_tick(maker_key);

        self.runner_books[runner_idx].decrement_level_total(maker_side, maker_tick, fill_amount);
        if maker_remaining.0 == 0 && self.orders.in_level_by_key(maker_key) {
            self.runner_books[runner_idx].unlink(&mut self.orders, maker_key, Money::zero());
        }
    }

    fn cmd_cancel_order(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        order_id: OrderId,
        account_id: AccountId,
        reason: &'static str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        let Some(order) = self.orders.get(&order_id) else {
            return Err(RejectReason::OrderNotFound);
        };

        if order.info.account_id != account_id {
            return Err(RejectReason::NotOrderOwner);
        }

        let is_live = matches!(
            order.info.state,
            BookOrderState::ExecutableUnmatched | BookOrderState::ExecutablePartiallyMatched
        );
        if !is_live {
            return Self::ok_noop(std::mem::take(&mut scratch.events));
        }

        scratch.events.push(self.emit(
            ts,
            BookEvent::OrderCancelled {
                order_id,
                reason: reason.to_string(),
            },
        ));
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    fn remove_from_levels(&mut self, order_id: OrderId) {
        if !self.orders.is_in_level(&order_id) {
            return;
        };

        let Some(order) = self.orders.get(&order_id) else {
            return;
        };

        let remaining = order.remaining();
        if remaining.0 <= 0 {
            return;
        }
        let Some(idx) = self.runner_index(order.runner_id) else {
            return;
        };
        let Some(key) = self.orders.get_key(&order_id) else {
            return;
        };
        self.runner_books[idx].unlink(&mut self.orders, key, remaining);
    }

    #[allow(clippy::too_many_arguments)]
    fn cmd_replace_order(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        old_order_id: OrderId,
        new_order_id: OrderId,
        account_id: AccountId,
        runner_id: RunnerId,
        side: Side,
        new_price: OddsX10000,
        new_stake: Money,
        persistence: Persistence,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        let Some(old) = self.orders.get(&old_order_id) else {
            return Err(RejectReason::OrderNotFound);
        };

        if old.info.account_id != account_id {
            return Err(RejectReason::NotOrderOwner);
        }

        let is_live = matches!(
            old.info.state,
            BookOrderState::ExecutableUnmatched | BookOrderState::ExecutablePartiallyMatched
        );
        if !is_live {
            return Err(RejectReason::OrderNotLive);
        }

        // Atomic: only cancel old if the replacement can be placed.
        let (place_events, place_ok) = self.cmd_place_order(
            scratch,
            ts,
            new_order_id,
            account_id,
            runner_id,
            side,
            new_price,
            new_stake,
            persistence,
            TimeInForce::Gtc,
        )?;

        let (cancel_events, _) =
            self.cmd_cancel_order(scratch, ts, old_order_id, account_id, "REPLACE")?;

        scratch.events.extend(cancel_events);
        scratch.events.extend(place_events);

        Ok((std::mem::take(&mut scratch.events), place_ok))
    }

    // ========================================================================
    // Runner Commands
    // ========================================================================

    fn cmd_remove_runner(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        runner_id: RunnerId,
        reduction_factor_bps: Option<u32>,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        let Some(_idx) = self.runner_index(runner_id) else {
            return Err(RejectReason::RunnerNotFound);
        };

        if self.removed_runners.contains(&runner_id) {
            return Err(RejectReason::RunnerAlreadyRemoved);
        }

        let orders_to_cancel: Vec<(OrderId, AccountId)> = self
            .orders
            .iter_sorted()
            .filter_map(|(oid, o)| {
                if o.runner_id == runner_id
                    && matches!(
                        o.info.state,
                        BookOrderState::ExecutableUnmatched
                            | BookOrderState::ExecutablePartiallyMatched
                    )
                {
                    Some((oid, o.info.account_id))
                } else {
                    None
                }
            })
            .collect();

        for (oid, account_id) in orders_to_cancel {
            let (cancel_events, _) =
                self.cmd_cancel_order(scratch, ts, oid, account_id, "RUNNER_REMOVED")?;
            scratch.events.extend(cancel_events);
        }

        let trades_to_void: Vec<TradeId> = self
            .trades
            .iter()
            .filter_map(|(&tid, t)| {
                if t.runner_id == runner_id && t.state == BookTradeState::Live {
                    Some(tid)
                } else {
                    None
                }
            })
            .collect();

        for tid in trades_to_void {
            scratch.events.push(self.emit(
                ts,
                BookEvent::TradeVoided {
                    trade_id: tid,
                    reason: "RUNNER_REMOVED".to_string(),
                },
            ));
        }

        scratch.events.push(self.emit(
            ts,
            BookEvent::RunnerRemoved {
                runner_id,
                reduction_factor_bps,
            },
        ));

        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    // ========================================================================
    // Settlement Commands
    // ========================================================================

    fn cmd_settle_market(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        runner_results: &[(RunnerId, RunnerResult)],
        dead_heat_divisor: Option<u32>,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        if self.state != BookMarketState::Closed {
            return Err(RejectReason::MarketNotClosed);
        }

        let result_runners: BTreeSet<_> = runner_results.iter().map(|(r, _)| *r).collect();
        if result_runners.len() != runner_results.len() {
            return Err(RejectReason::DuplicateRunner);
        }

        let active_runners: BTreeSet<_> = self
            .runner_ids
            .iter()
            .filter(|r| !self.removed_runners.contains(r))
            .copied()
            .collect();

        if result_runners != active_runners {
            return Err(RejectReason::IncompleteResults);
        }

        scratch.events.push(self.emit(
            ts,
            BookEvent::MarketSettled {
                runner_results: runner_results.to_vec(),
                dead_heat_divisor,
            },
        ));

        self.state_change(ts, BookMarketState::Settled, "SETTLED", &mut scratch.events);
        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    // ========================================================================
    // Trade Void Commands
    // ========================================================================

    fn cmd_void_trades_from_time(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        from_inclusive: DateTime,
        reason: &str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        if self.state == BookMarketState::Voided || self.state == BookMarketState::Settled {
            return Err(RejectReason::MarketTerminal);
        }

        let mut targets: Vec<(DateTime, TradeId)> = self
            .trades
            .iter()
            .filter_map(|(&tid, t)| {
                (t.state == BookTradeState::Live && t.matched_at >= from_inclusive)
                    .then_some((t.matched_at, tid))
            })
            .collect();
        targets.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1)));

        for (_, tid) in targets {
            scratch.events.push(self.emit(
                ts,
                BookEvent::TradeVoided {
                    trade_id: tid,
                    reason: reason.to_string(),
                },
            ));
        }

        Self::ok_noop(std::mem::take(&mut scratch.events))
    }

    fn cmd_void_trade_ids(
        &self,
        scratch: &mut Scratch,
        ts: DateTime,
        trade_ids: &[TradeId],
        reason: &str,
    ) -> Result<(EventVec, Option<CommandResponseKind>), RejectReason> {
        scratch.events.clear();

        if self.state == BookMarketState::Voided || self.state == BookMarketState::Settled {
            return Err(RejectReason::MarketTerminal);
        }

        let mut trade_ids: Vec<TradeId> = trade_ids.to_vec();
        trade_ids.sort();
        trade_ids.dedup();

        for tid in trade_ids {
            let Some(t) = self.trades.get(&tid) else {
                continue;
            };
            if t.state == BookTradeState::Voided {
                continue;
            }
            scratch.events.push(self.emit(
                ts,
                BookEvent::TradeVoided {
                    trade_id: tid,
                    reason: reason.to_string(),
                },
            ));
        }

        Self::ok_noop(std::mem::take(&mut scratch.events))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn replace_rejects_non_live_order() {
        let mid = MarketId(1);
        let ts = Utc::now();
        let mut book = TwoRunnerBook::new(mid, MarketKind::InPlayCapable, RunnerId(1), RunnerId(2));

        book.orders
            .insert(
                OrderId(1),
                BookOrder {
                    info: BookOrderInfo {
                        order_id: OrderId(1),
                        account_id: AccountId(10),
                        side: Side::Yes,
                        state: BookOrderState::Cancelled,
                        created_at: ts,
                        last_updated_at: ts,
                    },
                    runner_id: RunnerId(1),
                    price: OddsX10000(20000),
                    stake: Money(100),
                    matched: Money(0),
                    persistence: Persistence::Persist,
                },
            )
            .expect("order insert should succeed");

        let err = book
            .handle_command(&Command {
                correlation_id: CorrelationId(1),
                market_id: mid,
                kind: CommandKind::ReplaceOrder {
                    account_id: AccountId(10),
                    order_id: OrderId(1),
                    new_odds: Some(OddsX10000(20200)),
                    new_stake: Some(Money(100)),
                },
            })
            .expect_err("replace should reject non-live order");

        assert_eq!(err.reason, RejectReason::OrderNotLive);
    }
}