nautilus_betfair/stream/

parse.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Parsing utilities that convert Betfair stream messages into Nautilus domain models.
//!
//! MCM (Market Change Messages) are parsed into order book deltas, trade ticks,
//! and instrument status updates. OCM (Order Change Messages) are parsed into
//! order status reports and fill reports.

use ahash::{AHashMap, AHashSet};
use nautilus_core::UnixNanos;
use nautilus_model::{
    data::{
        BookOrder, InstrumentClose, InstrumentStatus, OrderBookDelta, OrderBookDeltas, TradeTick,
    },
    enums::{
        AggressorSide, BookAction, InstrumentCloseType, LiquiditySide, MarketStatusAction,
        OrderSide, OrderStatus, OrderType, RecordFlag, TimeInForce,
    },
    identifiers::{AccountId, ClientOrderId, InstrumentId, TradeId, VenueOrderId},
    reports::{FillReport, OrderStatusReport},
    types::{Currency, Money, Price, Quantity},
};
use rust_decimal::Decimal;

use crate::{
    common::{
        enums::{
            MarketStatus, RunnerStatus, StreamingOrderStatus, StreamingOrderType,
            resolve_streaming_order_status,
        },
        parse::{
            make_instrument_id, normalize_betfair_price, normalize_betfair_quantity,
            parse_betfair_price, parse_betfair_quantity, parse_millis_timestamp,
        },
    },
    data_types::{
        BetfairBspBookDelta, BetfairCricketMatch, BetfairRaceProgress, BetfairRaceRunnerData,
        BetfairStartingPrice, BetfairTicker,
    },
    stream::messages::{
        CricketChange, MarketDefinition, RaceProgressChange, RaceRunnerChange, RunnerChange,
        UnmatchedOrder,
    },
};

/// Parses a single runner's book data into [`OrderBookDeltas`].
///
/// Handles both full image snapshots (`is_snapshot = true`) and delta updates.
///
/// Only processes price-keyed fields (`atb`/`atl`). Level-indexed fields
/// (`batb`/`batl`/`bdatb`/`bdatl`) are ignored because correct delta
/// processing requires stateful level-to-price tracking. The stream
/// subscription should use `EX_ALL_OFFERS` which populates `atb`/`atl`.
///
/// Book side mapping (Betfair exchange convention):
/// - `atb` (available to back) -> [`OrderSide::Buy`] (bid side)
/// - `atl` (available to lay) -> [`OrderSide::Sell`] (ask side)
///
/// For snapshots: emits a Clear delta followed by Add deltas for each level.
/// For updates: emits Update or Delete (when volume is zero) deltas.
///
/// Returns `Ok(None)` if the runner change contains no processable book data.
///
/// # Errors
///
/// Returns an error if price or quantity values cannot be converted.
pub fn parse_runner_book_deltas(
    instrument_id: InstrumentId,
    rc: &RunnerChange,
    is_snapshot: bool,
    sequence: u64,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> anyhow::Result<Option<OrderBookDeltas>> {
    let atb_len = rc.atb.as_ref().map_or(0, Vec::len);
    let atl_len = rc.atl.as_ref().map_or(0, Vec::len);
    let total_levels = atb_len + atl_len;

    if total_levels == 0 && !is_snapshot {
        return Ok(None);
    }

    let snapshot_flags = if is_snapshot {
        RecordFlag::F_SNAPSHOT as u8
    } else {
        0
    };
    let mut deltas = Vec::with_capacity(total_levels + usize::from(is_snapshot));

    if is_snapshot {
        let mut clear = OrderBookDelta::clear(instrument_id, sequence, ts_event, ts_init);

        if total_levels == 0 {
            clear.flags |= RecordFlag::F_LAST as u8;
        }
        deltas.push(clear);
    }

    for (levels, side) in [
        (rc.atb.as_deref().unwrap_or(&[]), OrderSide::Buy),
        (rc.atl.as_deref().unwrap_or(&[]), OrderSide::Sell),
    ] {
        for pv in levels {
            if is_snapshot && pv.volume == Decimal::ZERO {
                continue;
            }

            let action = if is_snapshot {
                BookAction::Add
            } else if pv.volume == Decimal::ZERO {
                BookAction::Delete
            } else {
                BookAction::Update
            };

            deltas.push(OrderBookDelta::new(
                instrument_id,
                action,
                BookOrder::new(
                    side,
                    parse_betfair_price(pv.price)?,
                    parse_betfair_quantity(pv.volume)?,
                    0,
                ),
                snapshot_flags,
                sequence,
                ts_event,
                ts_init,
            ));
        }
    }

    // Set F_LAST on the final delta
    if let Some(last) = deltas.last_mut() {
        last.flags |= RecordFlag::F_LAST as u8;
    }

    Ok(Some(OrderBookDeltas::new(instrument_id, deltas)))
}

/// Creates a [`TradeTick`] from stream data.
///
/// Betfair does not identify the aggressor side in its stream, so
/// [`AggressorSide::NoAggressor`] is always used.
#[must_use]
pub fn make_trade_tick(
    instrument_id: InstrumentId,
    price: Price,
    size: Quantity,
    trade_id: TradeId,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> TradeTick {
    TradeTick::new(
        instrument_id,
        price,
        size,
        AggressorSide::NoAggressor,
        trade_id,
        ts_event,
        ts_init,
    )
}

/// Produces per-runner [`InstrumentStatus`] events from a market definition.
///
/// Iterates `def.runners` and maps each runner's lifecycle to a Nautilus status.
/// Scratched runners (`Removed`, `RemovedVacant`) close immediately regardless
/// of market-level state. The `in_play` flag distinguishes pre-open (Open + not
/// in play) from active trading (Open + in play).
///
/// Returns an empty vector when `def.status` or `def.runners` is missing.
#[must_use]
pub fn parse_instrument_statuses(
    market_id: &str,
    def: &MarketDefinition,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Vec<InstrumentStatus> {
    let Some(status) = def.status else {
        return Vec::new();
    };
    let Some(runners) = &def.runners else {
        return Vec::new();
    };
    let in_play = def.in_play.unwrap_or(false);

    if status == MarketStatus::Unknown {
        log::warn!("Skipping unmodeled Betfair market status for market {market_id}");
        return Vec::new();
    }

    runners
        .iter()
        .filter_map(|rd| {
            let handicap = rd.hc.unwrap_or(Decimal::ZERO);
            let instrument_id = make_instrument_id(market_id, rd.id, handicap);
            if rd.status == Some(RunnerStatus::Unknown) {
                log::warn!("Skipping unmodeled Betfair runner status for {instrument_id}");
                return None;
            }
            let action = match rd.status {
                Some(RunnerStatus::Removed | RunnerStatus::RemovedVacant) => {
                    MarketStatusAction::Close
                }
                _ => match (status, in_play) {
                    (MarketStatus::Inactive, _) => MarketStatusAction::Close,
                    (MarketStatus::Open, false) => MarketStatusAction::PreOpen,
                    (MarketStatus::Open, true) => MarketStatusAction::Trading,
                    (MarketStatus::Suspended, _) => MarketStatusAction::Pause,
                    (MarketStatus::Closed, _) => MarketStatusAction::Close,
                    // Unreachable: unmodeled market status skips the whole definition above.
                    (MarketStatus::Unknown, _) => MarketStatusAction::None,
                },
            };
            let is_trading = action == MarketStatusAction::Trading;
            Some(InstrumentStatus::new(
                instrument_id,
                action,
                ts_event,
                ts_init,
                None,
                None,
                Some(is_trading),
                None,
                None,
            ))
        })
        .collect()
}

/// Generates a deterministic [`TradeId`] for a Betfair fill.
///
/// Uses `bet_id` and cumulative `sm` (size matched) which together uniquely
/// identify each fill state, since `sm` increases monotonically with each fill.
pub fn make_trade_id(uo: &UnmatchedOrder) -> TradeId {
    let sm = normalize_betfair_quantity(uo.sm.unwrap_or(Decimal::ZERO));
    make_trade_id_for_size(&uo.id, sm)
}

/// Tracks cumulative fill state per bet to compute incremental fills from the
/// Betfair OCM stream.
///
/// Betfair provides cumulative `sm` (size matched) and `avp` (average price
/// matched) on each order update. This tracker maintains per-bet state to
/// derive individual fill quantities and prices for each update.
#[derive(Clone, Debug, Default)]
pub struct FillTracker {
    filled_qty: AHashMap<String, Decimal>,
    voided_qty: AHashMap<String, Decimal>,
    avg_px: AHashMap<String, Decimal>,
    published_trade_ids: AHashSet<String>,
    fill_lots: AHashMap<String, Vec<FillLot>>,
    fill_voids: AHashMap<(String, TradeId), Decimal>,
}

#[derive(Debug, Clone)]
struct FillLot {
    trade_id: TradeId,
    quantity: Decimal,
    price: Price,
    accounting_px: Price,
}

/// One cumulative per-fill allocation derived from Betfair's cumulative `sv`.
#[derive(Debug, Clone)]
pub struct FillVoidAllocation {
    pub trade_id: TradeId,
    pub voided_qty: Quantity,
    pub last_px: Price,
}

impl FillTracker {
    /// Creates a new [`FillTracker`] instance.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Computes a [`FillReport`] for an unmatched order update.
    ///
    /// A new fill is incremental. A price restatement repeats the surviving lot's trade ID
    /// when matched and voided quantities are unchanged, the average changed, and exactly one
    /// lot survives. Returns `None` when those conditions are not met, including a duplicate
    /// trade ID or an overfill.
    #[expect(clippy::too_many_arguments)]
    pub fn maybe_fill_report(
        &mut self,
        uo: &UnmatchedOrder,
        order_qty: Decimal,
        instrument_id: InstrumentId,
        account_id: AccountId,
        currency: Currency,
        ts_event: UnixNanos,
        ts_init: UnixNanos,
    ) -> Option<FillReport> {
        let raw_sm = uo.sm?;
        let sm = normalize_betfair_quantity(raw_sm);
        let order_qty = normalize_betfair_quantity(resolve_stream_order_quantity(order_qty, uo));

        if sm > order_qty {
            log::warn!(
                "Rejecting potential overfill for bet_id={}: order_qty={order_qty}, sm={sm}",
                uo.id,
            );
            return None;
        }

        let (trade_id, last_qty, last_px) =
            if uo.sv.is_some_and(|sv| sv > Decimal::ZERO) && self.has_fill_lots(&uo.id) {
                self.advance_cumulative_fill_with_voids(
                    &uo.id,
                    raw_sm,
                    uo.sv.unwrap_or(Decimal::ZERO),
                    uo.avp,
                    uo.p,
                )?
            } else {
                self.advance_cumulative_fill(&uo.id, raw_sm, uo.avp, uo.p)?
            };

        let venue_order_id = VenueOrderId::from(uo.id.as_str());
        let order_side = OrderSide::from(uo.side);
        let client_order_id = uo
            .rfo
            .as_deref()
            .filter(|s| !s.is_empty())
            .map(ClientOrderId::from);
        let ts_fill = uo.md.map_or(ts_event, parse_millis_timestamp);

        Some(make_fill_report(
            account_id,
            instrument_id,
            venue_order_id,
            trade_id,
            order_side,
            last_qty,
            last_px,
            currency,
            client_order_id,
            ts_fill,
            ts_init,
        ))
    }

    pub(crate) fn advance_cumulative_fill(
        &mut self,
        bet_id: &str,
        raw_size_matched: Decimal,
        average_price_matched: Option<Decimal>,
        fallback_price: Decimal,
    ) -> Option<(TradeId, Quantity, Price)> {
        self.advance_cumulative_fill_inner(
            bet_id,
            raw_size_matched,
            None,
            average_price_matched,
            fallback_price,
        )
    }

    pub(crate) fn advance_cumulative_fill_with_voids(
        &mut self,
        bet_id: &str,
        raw_size_matched: Decimal,
        cumulative_voided: Decimal,
        average_price_matched: Option<Decimal>,
        fallback_price: Decimal,
    ) -> Option<(TradeId, Quantity, Price)> {
        self.advance_cumulative_fill_inner(
            bet_id,
            raw_size_matched,
            Some(cumulative_voided),
            average_price_matched,
            fallback_price,
        )
    }

    fn advance_cumulative_fill_inner(
        &mut self,
        bet_id: &str,
        raw_size_matched: Decimal,
        cumulative_voided: Option<Decimal>,
        average_price_matched: Option<Decimal>,
        fallback_price: Decimal,
    ) -> Option<(TradeId, Quantity, Price)> {
        let size_matched = normalize_betfair_quantity(raw_size_matched);
        if size_matched <= Decimal::ZERO {
            return None;
        }

        let previous_filled = self
            .filled_qty
            .get(bet_id)
            .copied()
            .map_or(Decimal::ZERO, normalize_betfair_quantity);

        if size_matched == previous_filled {
            let restatement =
                self.matched_price_restatement(bet_id, cumulative_voided, average_price_matched);
            if let Some((_, _, last_px)) = restatement {
                self.set_surviving_accounting_px(bet_id, last_px);
            }

            self.record_cumulative_state(bet_id, size_matched, average_price_matched);
            return restatement;
        }

        if size_matched < previous_filled {
            return None;
        }

        let trade_id = make_trade_id_for_size(bet_id, size_matched);
        let raw_trade_id = make_trade_id_for_size(bet_id, raw_size_matched);
        if self.published_trade_ids.contains(trade_id.as_str())
            || self.published_trade_ids.contains(raw_trade_id.as_str())
        {
            self.record_cumulative_state(bet_id, size_matched, average_price_matched);
            self.published_trade_ids.insert(trade_id.to_string());
            return None;
        }

        let fill_qty = size_matched - previous_filled;
        let fill_price = if let Some(cumulative_voided) = cumulative_voided {
            self.compute_fill_price_with_voids(
                bet_id,
                average_price_matched,
                fallback_price,
                size_matched,
                fill_qty,
                cumulative_voided,
            )
        } else {
            self.compute_fill_price(
                bet_id,
                average_price_matched,
                fallback_price,
                previous_filled,
                size_matched,
            )
        };
        let last_qty = parse_betfair_quantity(fill_qty).ok()?;
        let last_px = parse_betfair_price(fill_price).ok()?;

        self.record_cumulative_state(bet_id, size_matched, average_price_matched);
        self.published_trade_ids.insert(trade_id.to_string());
        self.fill_lots
            .entry(bet_id.to_string())
            .or_default()
            .push(FillLot {
                trade_id,
                quantity: fill_qty,
                price: last_px,
                accounting_px: last_px,
            });

        Some((trade_id, last_qty, last_px))
    }

    /// Allocates cumulative Betfair `sv` to known applied fill lots, newest first.
    ///
    /// First-seen voids without known fill lots are recorded but emit no correction.
    pub fn maybe_fill_voids(&mut self, uo: &UnmatchedOrder) -> Vec<FillVoidAllocation> {
        let cumulative = normalize_betfair_quantity(uo.sv.unwrap_or(Decimal::ZERO));
        let previous = self
            .voided_qty
            .get(&uo.id)
            .copied()
            .unwrap_or(Decimal::ZERO);

        if cumulative <= previous {
            return Vec::new();
        }

        let Some(lots) = self.fill_lots.get(&uo.id) else {
            self.voided_qty.insert(uo.id.clone(), cumulative);
            return Vec::new();
        };
        let mut remaining = cumulative - previous;
        let mut desired = Vec::new();

        for lot in lots.iter().rev() {
            if remaining <= Decimal::ZERO {
                break;
            }
            let key = (uo.id.clone(), lot.trade_id);
            let prior = self.fill_voids.get(&key).copied().unwrap_or(Decimal::ZERO);
            let available = lot.quantity.saturating_sub(prior);
            let increment = remaining.min(available);
            if increment > Decimal::ZERO {
                desired.push((lot, prior + increment));
                remaining -= increment;
            }
        }

        if remaining > Decimal::ZERO {
            log::warn!(
                "Betfair cumulative void exceeds known fill lots for bet_id={}: sv={cumulative}, unmatched={remaining}",
                uo.id,
            );
            return Vec::new();
        }

        let mut updates = Vec::new();

        for (lot, allocation) in desired.into_iter().rev() {
            let key = (uo.id.clone(), lot.trade_id);
            let Ok(voided_qty) = parse_betfair_quantity(allocation) else {
                continue;
            };
            self.fill_voids.insert(key, allocation);
            updates.push(FillVoidAllocation {
                trade_id: lot.trade_id,
                voided_qty,
                last_px: lot.price,
            });
        }
        self.voided_qty.insert(uo.id.clone(), cumulative);
        updates
    }

    #[must_use]
    pub(crate) fn has_fill_lots(&self, bet_id: &str) -> bool {
        self.fill_lots
            .get(bet_id)
            .is_some_and(|lots| !lots.is_empty())
    }

    /// Returns whether a cumulative Betfair void update has not yet been applied.
    #[must_use]
    pub fn has_unseen_fill_void(&self, uo: &UnmatchedOrder) -> bool {
        let cumulative = normalize_betfair_quantity(uo.sv.unwrap_or(Decimal::ZERO));
        let previous = self
            .voided_qty
            .get(&uo.id)
            .copied()
            .unwrap_or(Decimal::ZERO);
        cumulative > previous
    }

    /// Returns whether an unchanged matched and voided quantity carries a new average while
    /// exactly one fill lot survives, without recording that average.
    #[must_use]
    pub(crate) fn has_pending_price_restatement(
        &self,
        bet_id: &str,
        size_matched: Decimal,
        size_voided: Decimal,
        average_price_matched: Option<Decimal>,
    ) -> bool {
        let size_voided = normalize_betfair_quantity(size_voided);
        let previous_voided = self
            .voided_qty
            .get(bet_id)
            .copied()
            .unwrap_or(Decimal::ZERO);

        if size_voided > previous_voided {
            return false;
        }

        let size_matched = normalize_betfair_quantity(size_matched);
        let cumulative = if self.has_fill_lots(bet_id) {
            size_matched + size_voided
        } else {
            size_matched
        };

        let previous = self
            .filled_qty
            .get(bet_id)
            .copied()
            .unwrap_or(Decimal::ZERO);

        if cumulative != previous || cumulative <= Decimal::ZERO {
            return false;
        }

        self.matched_price_restatement(bet_id, Some(size_voided), average_price_matched)
            .is_some()
    }

    /// Returns whether an update with unchanged matched and voided quantities carries a new
    /// average price while exactly one fill lot survives.
    #[must_use]
    pub(crate) fn has_unseen_price_restatement(&self, uo: &UnmatchedOrder) -> bool {
        self.has_pending_price_restatement(
            &uo.id,
            uo.sm.unwrap_or(Decimal::ZERO),
            uo.sv.unwrap_or(Decimal::ZERO),
            uo.avp,
        )
    }

    /// Returns whether a cumulative Betfair fill update has not yet been applied.
    #[must_use]
    pub fn has_unseen_fill(&self, uo: &UnmatchedOrder) -> bool {
        let size_matched = normalize_betfair_quantity(uo.sm.unwrap_or(Decimal::ZERO));
        let cumulative = if self.has_fill_lots(&uo.id) {
            size_matched + normalize_betfair_quantity(uo.sv.unwrap_or(Decimal::ZERO))
        } else {
            size_matched
        };
        let previous = self
            .filled_qty
            .get(&uo.id)
            .copied()
            .unwrap_or(Decimal::ZERO);
        let order_qty = normalize_betfair_quantity(resolve_stream_order_quantity(uo.s, uo));
        cumulative > previous && cumulative <= order_qty
    }

    pub(crate) fn sync_fill_lot(
        &mut self,
        bet_id: &str,
        trade_id: TradeId,
        quantity: Decimal,
        price: Price,
        voided_qty: Decimal,
    ) {
        let lots = self.fill_lots.entry(bet_id.to_string()).or_default();
        if let Some(lot) = lots.iter_mut().find(|lot| lot.trade_id == trade_id) {
            let stale_original = lot.accounting_px != price && price == lot.price;
            if !stale_original {
                lot.accounting_px = price;
            }
        } else {
            lots.push(FillLot {
                trade_id,
                quantity: normalize_betfair_quantity(quantity),
                price,
                accounting_px: price,
            });
        }
        let gross_filled = lots.iter().map(|lot| lot.quantity).sum();
        self.filled_qty.insert(bet_id.to_string(), gross_filled);
        self.published_trade_ids.insert(trade_id.to_string());
        if voided_qty > Decimal::ZERO {
            self.fill_voids.insert(
                (bet_id.to_string(), trade_id),
                normalize_betfair_quantity(voided_qty),
            );
        }
    }

    pub(crate) fn set_accounting_px(&mut self, bet_id: &str, trade_id: TradeId, price: Price) {
        let Some(lots) = self.fill_lots.get_mut(bet_id) else {
            return;
        };

        if let Some(lot) = lots.iter_mut().find(|lot| lot.trade_id == trade_id) {
            lot.accounting_px = price;
        }
    }

    pub(crate) fn matched_quantity(&self, bet_id: &str) -> Decimal {
        let filled = self.filled_qty.get(bet_id).copied().unwrap_or_default();
        let voided = self.voided_qty.get(bet_id).copied().unwrap_or_default();
        (filled - voided).max(Decimal::ZERO)
    }

    pub(crate) fn voided_quantity(&self, bet_id: &str) -> Decimal {
        self.voided_qty.get(bet_id).copied().unwrap_or_default()
    }

    pub(crate) fn sync_voided_qty(&mut self, bet_id: &str, voided_qty: Decimal) {
        self.voided_qty
            .insert(bet_id.to_string(), normalize_betfair_quantity(voided_qty));
    }

    fn matched_price_restatement(
        &self,
        bet_id: &str,
        cumulative_voided: Option<Decimal>,
        average_price_matched: Option<Decimal>,
    ) -> Option<(TradeId, Quantity, Price)> {
        if !self.size_voided_unchanged(bet_id, cumulative_voided) {
            return None;
        }

        let avp = average_price_matched.map(normalize_betfair_price)?;
        let prev_avg = self.avg_px.get(bet_id).copied()?;
        if avp == prev_avg {
            return None;
        }

        let (trade_id, surviving_qty) = self.single_surviving_lot(bet_id)?;

        let Ok(last_qty) = parse_betfair_quantity(surviving_qty) else {
            log::warn!("Cannot restate matched quantity {surviving_qty} for bet_id={bet_id}");
            return None;
        };

        let Ok(last_px) = parse_betfair_price(avp) else {
            log::warn!("Cannot restate matched price {avp} for bet_id={bet_id}");
            return None;
        };

        Some((trade_id, last_qty, last_px))
    }

    fn set_surviving_accounting_px(&mut self, bet_id: &str, price: Price) {
        let Some(lots) = self.fill_lots.get_mut(bet_id) else {
            return;
        };

        for lot in lots.iter_mut() {
            let voided = self
                .fill_voids
                .get(&(bet_id.to_string(), lot.trade_id))
                .copied()
                .unwrap_or(Decimal::ZERO);

            if lot.quantity.saturating_sub(voided) > Decimal::ZERO {
                lot.accounting_px = price;
            }
        }
    }

    fn size_voided_unchanged(&self, bet_id: &str, cumulative_voided: Option<Decimal>) -> bool {
        let current = cumulative_voided.map_or(Decimal::ZERO, normalize_betfair_quantity);
        let previous = self
            .voided_qty
            .get(bet_id)
            .copied()
            .unwrap_or(Decimal::ZERO);
        current == previous
    }

    pub(crate) fn seed_missing_averages(&mut self) {
        let bet_ids = self.fill_lots.keys().cloned().collect::<Vec<_>>();
        for bet_id in bet_ids {
            if self.avg_px.contains_key(&bet_id) {
                continue;
            }

            let Some(lots) = self.fill_lots.get(&bet_id) else {
                continue;
            };

            let mut quantity = Decimal::ZERO;
            let mut notional = Decimal::ZERO;

            for lot in lots {
                let voided = self
                    .fill_voids
                    .get(&(bet_id.clone(), lot.trade_id))
                    .copied()
                    .unwrap_or(Decimal::ZERO);
                let surviving = lot.quantity.saturating_sub(voided);
                if surviving <= Decimal::ZERO {
                    continue;
                }

                quantity += surviving;
                notional += surviving * lot.accounting_px.as_decimal();
            }

            if quantity > Decimal::ZERO {
                self.avg_px.insert(bet_id, notional / quantity);
            }
        }
    }

    fn single_surviving_lot(&self, bet_id: &str) -> Option<(TradeId, Decimal)> {
        let lots = self.fill_lots.get(bet_id)?;

        let mut surviving = lots.iter().filter_map(|lot| {
            let voided = self
                .fill_voids
                .get(&(bet_id.to_string(), lot.trade_id))
                .copied()
                .unwrap_or(Decimal::ZERO);
            let surviving_qty = lot.quantity.saturating_sub(voided);
            (surviving_qty > Decimal::ZERO).then_some((lot.trade_id, surviving_qty))
        });

        let lot = surviving.next()?;
        surviving.next().is_none().then_some(lot)
    }

    fn record_cumulative_state(
        &mut self,
        bet_id: &str,
        size_matched: Decimal,
        average_price_matched: Option<Decimal>,
    ) {
        self.filled_qty.insert(bet_id.to_string(), size_matched);
        if let Some(avg_px) = average_price_matched.map(normalize_betfair_price) {
            self.avg_px.insert(bet_id.to_string(), avg_px);
        }
    }

    /// Back-calculates the individual fill price from Betfair's cumulative
    /// average price matched (`avp`).
    ///
    /// For the first fill, the average price IS the fill price. For subsequent
    /// fills, the individual price is derived from:
    /// `fill_price = (avp * sm - prev_avp * prev_sm) / fill_size`
    fn compute_fill_price(
        &self,
        bet_id: &str,
        average_price_matched: Option<Decimal>,
        fallback_price: Decimal,
        prev_filled: Decimal,
        sm: Decimal,
    ) -> Decimal {
        let Some(avp) = average_price_matched.map(normalize_betfair_price) else {
            return fallback_price;
        };

        if prev_filled == Decimal::ZERO {
            return avp;
        }

        let Some(prev_avg) = self.avg_px.get(bet_id).copied() else {
            return avp;
        };

        if prev_avg == avp {
            return avp;
        }

        let fill_size = sm - prev_filled;

        if fill_size == Decimal::ZERO {
            return prev_avg;
        }

        let fill_price = (avp * sm - prev_avg * prev_filled) / fill_size;

        if fill_price <= Decimal::ZERO {
            log::warn!(
                "Calculated fill price {fill_price} is invalid for bet_id={bet_id}, falling back to avp={avp}",
            );
            return avp;
        }

        fill_price
    }

    fn compute_fill_price_with_voids(
        &self,
        bet_id: &str,
        average_price_matched: Option<Decimal>,
        fallback_price: Decimal,
        gross_matched: Decimal,
        fill_qty: Decimal,
        cumulative_voided: Decimal,
    ) -> Decimal {
        let Some(avp) = average_price_matched.map(normalize_betfair_price) else {
            return fallback_price;
        };
        let cumulative_voided = normalize_betfair_quantity(cumulative_voided);
        let previous_voided = self
            .voided_qty
            .get(bet_id)
            .copied()
            .map_or(Decimal::ZERO, normalize_betfair_quantity);
        let incremental_void = cumulative_voided.saturating_sub(previous_voided);
        let surviving_fill_qty = fill_qty.saturating_sub(incremental_void);
        if surviving_fill_qty <= Decimal::ZERO {
            return avp;
        }

        let mut prior_notional = Decimal::ZERO;
        let mut effective_lots = Vec::new();

        if let Some(lots) = self.fill_lots.get(bet_id) {
            for lot in lots {
                let already_voided = self
                    .fill_voids
                    .get(&(bet_id.to_string(), lot.trade_id))
                    .copied()
                    .unwrap_or(Decimal::ZERO);
                let effective = lot.quantity.saturating_sub(already_voided);
                prior_notional += effective * lot.accounting_px.as_decimal();
                effective_lots.push((effective, lot.accounting_px.as_decimal()));
            }
        }

        let mut prior_void = incremental_void.saturating_sub(fill_qty);
        for (effective, price) in effective_lots.into_iter().rev() {
            if prior_void <= Decimal::ZERO {
                break;
            }
            let removed = prior_void.min(effective);
            prior_notional -= removed * price;
            prior_void -= removed;
        }

        let surviving_matched = gross_matched.saturating_sub(cumulative_voided);
        let fill_price = (avp * surviving_matched - prior_notional) / surviving_fill_qty;
        if fill_price <= Decimal::ZERO {
            log::warn!(
                "Calculated post-void fill price {fill_price} is invalid for bet_id={bet_id}, falling back to avp={avp}",
            );
            return avp;
        }
        fill_price
    }

    /// Pre-populates state for a bet from existing order data.
    ///
    /// Monotonic in `filled_qty`: keeps the larger of the cached and
    /// in-memory values so a cache that lags the tracker (engine has
    /// not yet processed an emitted fill) cannot regress it.
    pub fn sync_order(&mut self, bet_id: &str, filled_qty: Decimal, avg_px: Decimal) {
        let filled_qty = normalize_betfair_quantity(filled_qty);
        let avg_px = normalize_betfair_price(avg_px);
        let current = self
            .filled_qty
            .get(bet_id)
            .copied()
            .map_or(Decimal::ZERO, normalize_betfair_quantity);

        if filled_qty > current {
            self.filled_qty.insert(bet_id.to_string(), filled_qty);
            if avg_px > Decimal::ZERO {
                self.avg_px.insert(bet_id.to_string(), avg_px);
            }
        }
    }

    /// Seeds the trade-id dedup set so fills already published via another
    /// channel are not re-emitted when the post-reconnect image arrives.
    pub fn seed_published_trade_ids<I, S>(&mut self, trade_ids: I)
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        for id in trade_ids {
            self.published_trade_ids.insert(id.into());
        }
    }

    /// Removes state for a completed bet to prevent unbounded growth.
    pub fn prune(&mut self, bet_id: &str) {
        self.filled_qty.remove(bet_id);
        self.voided_qty.remove(bet_id);
        self.avg_px.remove(bet_id);
        self.fill_lots.remove(bet_id);
        self.fill_voids.retain(|(id, _), _| id != bet_id);

        let prefix = format!("{bet_id}-");
        self.published_trade_ids
            .retain(|id| !id.starts_with(&prefix));
    }
}

fn make_trade_id_for_size(bet_id: &str, size_matched: Decimal) -> TradeId {
    TradeId::new(format!("{bet_id}-{size_matched}"))
}

/// Returns `true` if the unmatched order has cancel, lapse, or void quantities.
#[must_use]
pub fn has_cancel_quantity(uo: &UnmatchedOrder) -> bool {
    let sc = uo.sc.unwrap_or(Decimal::ZERO);
    let sl = uo.sl.unwrap_or(Decimal::ZERO);
    let sv = uo.sv.unwrap_or(Decimal::ZERO);
    (sc + sl + sv) > Decimal::ZERO
}

/// Returns `true` if the order is execution-complete and has lapsed.
#[must_use]
pub fn is_lapsed(uo: &UnmatchedOrder) -> bool {
    uo.status == StreamingOrderStatus::ExecutionComplete && uo.lsrc.is_some()
}

/// Returns `true` if the unmatched order is an SP (on-close) bet resting until
/// BSP reconciliation: execution-complete with no matched, cancelled, lapsed,
/// or voided quantity and no lapse reason.
///
/// SP bets cannot be cancelled once placed, so this state is open, not
/// terminal.
#[must_use]
pub fn is_resting_sp_bet(uo: &UnmatchedOrder) -> bool {
    uo.status == StreamingOrderStatus::ExecutionComplete
        && matches!(
            uo.ot,
            StreamingOrderType::LimitOnClose | StreamingOrderType::MarketOnClose
        )
        && uo.sm.unwrap_or(Decimal::ZERO) <= Decimal::ZERO
        && uo.lsrc.is_none()
        && !has_cancel_quantity(uo)
}

/// Parses a streaming [`UnmatchedOrder`] into a Nautilus [`OrderStatusReport`].
///
/// Resolves the Nautilus order status from the Betfair streaming status
/// plus matched/cancelled quantities.
///
/// # Errors
///
/// Returns an error if price or quantity values cannot be converted.
pub fn parse_order_status_report(
    uo: &UnmatchedOrder,
    instrument_id: InstrumentId,
    account_id: AccountId,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let order_side = OrderSide::from(uo.side);
    let order_type = OrderType::from(uo.ot);
    let time_in_force = parse_stream_time_in_force(uo)?;

    let size_matched = uo.sm.unwrap_or(Decimal::ZERO);
    let size_cancelled = uo.sc.unwrap_or(Decimal::ZERO);
    let size_lapsed = uo.sl.unwrap_or(Decimal::ZERO);
    let size_voided = uo.sv.unwrap_or(Decimal::ZERO);

    // Include lapsed/voided in the closed quantity for status resolution
    let size_closed = size_cancelled + size_lapsed + size_voided;
    let order_status = if uo.status == StreamingOrderStatus::ExecutionComplete
        && size_voided > Decimal::ZERO
        && size_cancelled.is_zero()
        && size_lapsed.is_zero()
    {
        OrderStatus::Voided
    } else if is_resting_sp_bet(uo) {
        OrderStatus::Accepted
    } else {
        resolve_streaming_order_status(uo.status, size_matched, size_closed)
    };

    let quantity_decimal = stream_order_quantity(uo);
    anyhow::ensure!(
        quantity_decimal > Decimal::ZERO,
        "failed to resolve positive quantity for stream order update {} \
         (order_type={:?}, persistence_type={:?}, size={}, bsp_liability={:?}, \
         size_matched={:?}, size_remaining={:?}, size_cancelled={:?}, size_lapsed={:?}, size_voided={:?})",
        uo.id,
        uo.ot,
        uo.pt,
        uo.s,
        uo.bsp,
        uo.sm,
        uo.sr,
        uo.sc,
        uo.sl,
        uo.sv,
    );
    let quantity = parse_betfair_quantity(quantity_decimal)?;
    let filled_qty = parse_betfair_quantity(size_matched)?;

    let ts_accepted = parse_millis_timestamp(uo.pd);

    // Use the latest lifecycle timestamp, falling back to OCM publish time
    let ts_last = [uo.md, uo.cd, uo.ld]
        .into_iter()
        .flatten()
        .max()
        .map_or(ts_event, parse_millis_timestamp);

    let venue_order_id = VenueOrderId::from(uo.id.as_str());
    let client_order_id = uo
        .rfo
        .as_deref()
        .filter(|s| !s.is_empty())
        .map(ClientOrderId::from);

    let price = parse_betfair_price(uo.p)?;

    let mut report = OrderStatusReport::new(
        account_id,
        instrument_id,
        client_order_id,
        venue_order_id,
        order_side.into(),
        order_type,
        time_in_force,
        order_status,
        quantity,
        filled_qty,
        ts_accepted,
        ts_last,
        ts_init,
        None,
    )
    .with_price(price);

    report.avg_px = uo.avp;
    if let Some(lsrc) = uo.lsrc {
        report.cancel_reason = Some(lsrc.to_string());
    }

    Ok(report)
}

fn parse_stream_time_in_force(uo: &UnmatchedOrder) -> anyhow::Result<TimeInForce> {
    if matches!(
        uo.ot,
        StreamingOrderType::LimitOnClose | StreamingOrderType::MarketOnClose
    ) {
        // The stream can carry a non-BSP persistence type on SP bets; the
        // on-close instruction defines the time in force
        return Ok(TimeInForce::AtTheClose);
    }

    match uo.pt {
        Some(persistence_type) => Ok(TimeInForce::from(persistence_type)),
        None => anyhow::bail!("missing persistence type for order update {}", uo.id),
    }
}

fn stream_order_quantity(uo: &UnmatchedOrder) -> Decimal {
    if uo.s > Decimal::ZERO {
        return uo.s;
    }

    let lifecycle_qty = uo.sm.unwrap_or(Decimal::ZERO)
        + uo.sr.unwrap_or(Decimal::ZERO)
        + uo.sc.unwrap_or(Decimal::ZERO)
        + uo.sl.unwrap_or(Decimal::ZERO)
        + uo.sv.unwrap_or(Decimal::ZERO);

    if lifecycle_qty > Decimal::ZERO {
        return lifecycle_qty;
    }

    if uses_liability_based_stream_quantity(uo) {
        return uo.bsp.unwrap_or(Decimal::ZERO);
    }

    Decimal::ZERO
}

fn resolve_stream_order_quantity(order_qty: Decimal, uo: &UnmatchedOrder) -> Decimal {
    if order_qty > Decimal::ZERO {
        order_qty
    } else {
        stream_order_quantity(uo)
    }
}

fn uses_liability_based_stream_quantity(uo: &UnmatchedOrder) -> bool {
    matches!(
        uo.ot,
        crate::common::enums::StreamingOrderType::LimitOnClose
            | crate::common::enums::StreamingOrderType::MarketOnClose
    )
}

/// Creates a [`FillReport`] for a Betfair order fill.
///
/// Betfair charges commission on net winnings, not per-fill, so commission
/// is set to zero. The `liquidity_side` is unknown from the stream.
#[must_use]
#[expect(clippy::too_many_arguments)]
pub fn make_fill_report(
    account_id: AccountId,
    instrument_id: InstrumentId,
    venue_order_id: VenueOrderId,
    trade_id: TradeId,
    order_side: OrderSide,
    last_qty: Quantity,
    last_px: Price,
    currency: Currency,
    client_order_id: Option<ClientOrderId>,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> FillReport {
    FillReport::new(
        account_id,
        instrument_id,
        venue_order_id,
        trade_id,
        order_side,
        last_qty,
        last_px,
        Money::zero(currency),
        LiquiditySide::NoLiquiditySide,
        client_order_id,
        None,
        ts_event,
        ts_init,
        None,
    )
}

/// Extracts a [`BetfairTicker`] from a runner change if any ticker fields are present.
///
/// Returns `None` when the runner change contains no ltp, tv, spn, or spf data.
#[must_use]
pub fn parse_betfair_ticker(
    instrument_id: InstrumentId,
    rc: &RunnerChange,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Option<BetfairTicker> {
    if rc.ltp.is_none() && rc.tv.is_none() && rc.spn.is_none() && rc.spf.is_none() {
        return None;
    }

    Some(BetfairTicker::new(
        instrument_id,
        rc.ltp,
        rc.tv,
        rc.spn,
        rc.spf,
        ts_event,
        ts_init,
    ))
}

/// Extracts [`BetfairStartingPrice`] values from a market definition's runners.
///
/// Returns one entry per runner that has a non-None BSP value.
#[must_use]
pub fn parse_betfair_starting_prices(
    market_id: &str,
    def: &MarketDefinition,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Vec<BetfairStartingPrice> {
    let Some(runners) = &def.runners else {
        return Vec::new();
    };

    runners
        .iter()
        .filter_map(|rd| {
            let bsp = rd.bsp?;
            let handicap = rd.hc.unwrap_or(Decimal::ZERO);
            let instrument_id = make_instrument_id(market_id, rd.id, handicap);
            Some(BetfairStartingPrice::new(
                instrument_id,
                bsp,
                ts_event,
                ts_init,
            ))
        })
        .collect()
}

/// Extracts BSP order book deltas from a runner change's `spb`/`spl` fields.
///
/// Returns an empty vec when neither `spb` nor `spl` data is present.
/// The `side` field uses `OrderSide::Sell` for `spb` (back) and
/// `OrderSide::Buy` for `spl` (lay), following Betfair's inverted convention.
#[must_use]
pub fn parse_bsp_book_deltas(
    instrument_id: InstrumentId,
    rc: &RunnerChange,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Vec<BetfairBspBookDelta> {
    let spb_len = rc.spb.as_ref().map_or(0, Vec::len);
    let spl_len = rc.spl.as_ref().map_or(0, Vec::len);

    if spb_len + spl_len == 0 {
        return Vec::new();
    }

    let mut result = Vec::with_capacity(spb_len + spl_len);

    for (levels, side) in [
        (rc.spb.as_deref().unwrap_or(&[]), OrderSide::Sell),
        (rc.spl.as_deref().unwrap_or(&[]), OrderSide::Buy),
    ] {
        for pv in levels {
            let action = if pv.volume == Decimal::ZERO {
                BookAction::Delete
            } else {
                BookAction::Update
            };

            result.push(BetfairBspBookDelta::new(
                instrument_id,
                action,
                side,
                pv.price,
                pv.volume,
                ts_event,
                ts_init,
            ));
        }
    }

    result
}

/// Produces [`InstrumentClose`] events from a market definition's runner statuses.
///
/// Winners and placed runners get close price 1.0; losers and removed runners
/// get close price 0.0. Active runners produce no close event.
#[must_use]
pub fn parse_instrument_closes(
    market_id: &str,
    def: &MarketDefinition,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Vec<InstrumentClose> {
    let Some(runners) = &def.runners else {
        return Vec::new();
    };

    runners
        .iter()
        .filter_map(|rd| {
            let status = rd.status.as_ref()?;
            let close_price = match status {
                RunnerStatus::Winner | RunnerStatus::Placed => Price::from("1.00"),
                RunnerStatus::Loser | RunnerStatus::Removed | RunnerStatus::RemovedVacant => {
                    Price::from("0.00")
                }
                RunnerStatus::Active | RunnerStatus::Hidden | RunnerStatus::Unknown => return None,
            };

            let handicap = rd.hc.unwrap_or(Decimal::ZERO);
            let instrument_id = make_instrument_id(market_id, rd.id, handicap);

            Some(InstrumentClose::new(
                instrument_id,
                close_price,
                InstrumentCloseType::ContractExpired,
                ts_event,
                ts_init,
            ))
        })
        .collect()
}

/// Parses a single [`RaceRunnerChange`] into a [`BetfairRaceRunnerData`].
///
/// Returns `None` if the runner change has no selection ID.
#[must_use]
pub fn parse_race_runner_data(
    race_id: &str,
    market_id: &str,
    rrc: &RaceRunnerChange,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Option<BetfairRaceRunnerData> {
    let selection_id = rrc.id?;

    Some(BetfairRaceRunnerData::new(
        race_id.to_string(),
        market_id.to_string(),
        selection_id,
        rrc.lat.unwrap_or(f64::NAN),
        rrc.lng.unwrap_or(f64::NAN),
        rrc.spd.unwrap_or(f64::NAN),
        rrc.prg.unwrap_or(f64::NAN),
        rrc.sfq.unwrap_or(f64::NAN),
        ts_event,
        ts_init,
    ))
}

/// Parses a [`RaceProgressChange`] into a [`BetfairRaceProgress`].
#[must_use]
pub fn parse_race_progress(
    race_id: &str,
    market_id: &str,
    rpc: &RaceProgressChange,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> BetfairRaceProgress {
    let order_json = serialize_json_value(rpc.ord.as_ref());
    let jumps_json = serialize_json_value(rpc.jumps.as_ref());

    BetfairRaceProgress::new(
        race_id.to_string(),
        market_id.to_string(),
        rpc.g.clone().unwrap_or_default(),
        rpc.st.unwrap_or(f64::NAN),
        rpc.rt.unwrap_or(f64::NAN),
        rpc.spd.unwrap_or(f64::NAN),
        rpc.prg.unwrap_or(f64::NAN),
        order_json,
        jumps_json,
        ts_event,
        ts_init,
    )
}

#[must_use]
pub fn parse_cricket_match(
    cricket: &CricketChange,
    ts_event: UnixNanos,
    ts_init: UnixNanos,
) -> Option<BetfairCricketMatch> {
    let event_id = cricket.event_id.clone()?;
    let market_id = cricket.market_id.clone()?;

    Some(BetfairCricketMatch::new(
        event_id,
        market_id,
        serialize_json_value(cricket.fixture_info.as_ref()),
        serialize_json_value(cricket.home_team.as_ref()),
        serialize_json_value(cricket.away_team.as_ref()),
        serialize_json_value(cricket.match_stats.as_ref()),
        serialize_json_value(cricket.incident_list_wrapper.as_ref()),
        ts_event,
        ts_init,
    ))
}

fn serialize_json_value<T>(value: Option<&T>) -> String
where
    T: serde::Serialize,
{
    value
        .map(|value| serde_json::to_string(value).unwrap_or_default())
        .unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use nautilus_model::enums::{MarketStatusAction, OrderStatus, TimeInForce};
    use rstest::rstest;

    use super::*;
    use crate::{
        common::{
            consts::{BETFAIR_PRICE_PRECISION, BETFAIR_QUANTITY_PRECISION},
            enums::{StreamingOrderType, StreamingPersistenceType, StreamingSide},
            testing::load_test_json,
        },
        stream::messages::{PV, RunnerChange, RunnerDefinition, StreamMessage, stream_decode},
    };

    fn assert_decimal_option_eq(actual: Option<Decimal>, expected: Decimal) {
        assert_eq!(actual, Some(expected));
    }

    #[rstest]
    fn test_parse_runner_book_snapshot() {
        let data = load_test_json("stream/mcm_live_IMAGE.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::MarketChange(mcm) = msg {
            let mc = mcm.mc.as_ref().unwrap();
            let change = &mc[0];
            let rc = &change.rc.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&change.id, rc.id, Decimal::ZERO);

            let deltas = parse_runner_book_deltas(
                instrument_id,
                rc,
                true,
                mcm.pt,
                parse_millis_timestamp(mcm.pt),
                parse_millis_timestamp(mcm.pt),
            )
            .unwrap()
            .expect("should produce deltas");

            // Clear + atb levels + atl levels
            let atb_len = rc.atb.as_ref().unwrap().len();
            let atl_len = rc.atl.as_ref().unwrap().len();
            assert_eq!(deltas.deltas.len(), 1 + atb_len + atl_len);

            // First delta is Clear with F_SNAPSHOT
            assert_eq!(deltas.deltas[0].action, BookAction::Clear);
            assert!(RecordFlag::F_SNAPSHOT.matches(deltas.deltas[0].flags));

            // Subsequent deltas are Add with F_SNAPSHOT
            for delta in &deltas.deltas[1..] {
                assert_eq!(delta.action, BookAction::Add);
                assert!(RecordFlag::F_SNAPSHOT.matches(delta.flags));
            }

            // Last delta has F_LAST
            let last = deltas.deltas.last().unwrap();
            assert!(RecordFlag::F_LAST.matches(last.flags));

            // Verify buy/sell sides
            let buy_count = deltas
                .deltas
                .iter()
                .filter(|d| d.order.side == Some(OrderSide::Buy))
                .count();
            let sell_count = deltas
                .deltas
                .iter()
                .filter(|d| d.order.side == Some(OrderSide::Sell))
                .count();
            assert_eq!(buy_count, atb_len);
            assert_eq!(sell_count, atl_len);
        } else {
            panic!("expected MarketChange");
        }
    }

    #[rstest]
    fn test_parse_runner_book_snapshot_skips_zero_volume_levels() {
        let rc: RunnerChange = serde_json::from_str(
            r#"{
                "id": 123,
                "atb": [[2.0, 0.0], [2.1, 3.0]],
                "atl": [[2.2, 0.0], [2.3, 4.0]]
            }"#,
        )
        .unwrap();
        let instrument_id = make_instrument_id("1.234", rc.id, Decimal::ZERO);

        let deltas = parse_runner_book_deltas(
            instrument_id,
            &rc,
            true,
            1_551_400_000_000,
            parse_millis_timestamp(1_551_400_000_000),
            parse_millis_timestamp(1_551_400_000_000),
        )
        .unwrap()
        .expect("should produce deltas");

        assert_eq!(deltas.deltas.len(), 3);
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
        assert!(
            deltas
                .deltas
                .iter()
                .filter(|delta| delta.action == BookAction::Add)
                .all(|delta| !delta.order.size.is_zero())
        );
    }

    #[rstest]
    fn test_parse_runner_book_update() {
        let data = load_test_json("stream/mcm_UPDATE.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::MarketChange(mcm) = msg {
            let mc = mcm.mc.as_ref().unwrap();
            let change = &mc[0];
            let rc = &change.rc.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&change.id, rc.id, Decimal::ZERO);

            let deltas = parse_runner_book_deltas(
                instrument_id,
                rc,
                false,
                mcm.pt,
                parse_millis_timestamp(mcm.pt),
                parse_millis_timestamp(mcm.pt),
            )
            .unwrap()
            .expect("should produce deltas");

            // No Clear delta for updates
            assert!(deltas.deltas.iter().all(|d| d.action != BookAction::Clear));

            // Last delta has F_LAST
            let last = deltas.deltas.last().unwrap();
            assert!(RecordFlag::F_LAST.matches(last.flags));

            // No snapshot flags
            for delta in &deltas.deltas {
                assert!(!RecordFlag::F_SNAPSHOT.matches(delta.flags));
            }
        } else {
            panic!("expected MarketChange");
        }
    }

    #[rstest]
    fn test_parse_runner_book_update_zero_volume_is_delete() {
        let data = load_test_json("stream/mcm_UPDATE.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::MarketChange(mcm) = msg {
            let mc = mcm.mc.as_ref().unwrap();
            let change = &mc[0];
            let rc = &change.rc.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&change.id, rc.id, Decimal::ZERO);

            let deltas = parse_runner_book_deltas(
                instrument_id,
                rc,
                false,
                mcm.pt,
                parse_millis_timestamp(mcm.pt),
                parse_millis_timestamp(mcm.pt),
            )
            .unwrap()
            .unwrap();

            // atl has [[4.7, 0]] which should be Delete
            assert!(
                deltas.deltas.iter().any(|d| d.action == BookAction::Delete),
                "zero volume should produce Delete action"
            );
        } else {
            panic!("expected MarketChange");
        }
    }

    #[rstest]
    fn test_parse_runner_book_no_levels_returns_none() {
        let rc = RunnerChange {
            id: 12345,
            hc: None,
            atb: None,
            atl: None,
            batb: None,
            batl: None,
            bdatb: None,
            bdatl: None,
            spb: None,
            spl: None,
            spn: None,
            spf: None,
            trd: None,
            ltp: None,
            tv: None,
        };

        let result = parse_runner_book_deltas(
            make_instrument_id("1.123", 12345, Decimal::ZERO),
            &rc,
            false,
            0,
            UnixNanos::default(),
            UnixNanos::default(),
        );

        assert!(result.unwrap().is_none());
    }

    #[rstest]
    fn test_make_trade_tick() {
        let instrument_id = make_instrument_id("1.180737206", 19248890, Decimal::ZERO);
        let tick = make_trade_tick(
            instrument_id,
            Price::new(2.42, BETFAIR_PRICE_PRECISION),
            Quantity::new(100.0, BETFAIR_QUANTITY_PRECISION),
            TradeId::from("test-trade-1"),
            UnixNanos::default(),
            UnixNanos::default(),
        );

        assert_eq!(tick.instrument_id, instrument_id);
        assert_eq!(tick.price.as_f64(), 2.42);
        assert_eq!(tick.size.as_f64(), 100.0);
        assert_eq!(tick.aggressor_side, AggressorSide::NoAggressor);
    }

    fn make_status_def(
        status: MarketStatus,
        in_play: bool,
        runner_status: RunnerStatus,
    ) -> MarketDefinition {
        MarketDefinition {
            runners: Some(vec![RunnerDefinition {
                id: 456,
                hc: None,
                sort_priority: None,
                name: None,
                status: Some(runner_status),
                adjustment_factor: None,
                bsp: None,
                removal_date: None,
            }]),
            bet_delay: None,
            betting_type: None,
            bsp_market: None,
            bsp_reconciled: None,
            competition_id: None,
            competition_name: None,
            complete: None,
            country_code: None,
            cross_matching: None,
            discount_allowed: None,
            each_way_divisor: None,
            event_id: None,
            event_name: None,
            event_type_id: None,
            event_type_name: None,
            in_play: Some(in_play),
            line_interval: None,
            line_max_unit: None,
            line_min_unit: None,
            market_base_rate: None,
            market_id: None,
            market_name: None,
            market_time: None,
            market_type: None,
            number_of_active_runners: None,
            number_of_winners: None,
            open_date: None,
            persistence_enabled: None,
            price_ladder_definition: None,
            race_type: None,
            regulators: None,
            runners_voidable: None,
            settled_time: None,
            status: Some(status),
            suspend_time: None,
            timezone: None,
            turn_in_play_enabled: None,
            venue: None,
            version: None,
        }
    }

    #[rstest]
    #[case(MarketStatus::Open, false, MarketStatusAction::PreOpen, false)]
    #[case(MarketStatus::Open, true, MarketStatusAction::Trading, true)]
    #[case(MarketStatus::Closed, false, MarketStatusAction::Close, false)]
    #[case(MarketStatus::Closed, true, MarketStatusAction::Close, false)]
    #[case(MarketStatus::Suspended, false, MarketStatusAction::Pause, false)]
    #[case(MarketStatus::Suspended, true, MarketStatusAction::Pause, false)]
    #[case(MarketStatus::Inactive, false, MarketStatusAction::Close, false)]
    fn test_parse_instrument_statuses_market_state(
        #[case] status: MarketStatus,
        #[case] in_play: bool,
        #[case] expected_action: MarketStatusAction,
        #[case] expected_is_trading: bool,
    ) {
        let def = make_status_def(status, in_play, RunnerStatus::Active);
        let results =
            parse_instrument_statuses("1.123", &def, UnixNanos::default(), UnixNanos::default());

        assert_eq!(results.len(), 1);
        assert_eq!(results[0].action, expected_action);
        assert_eq!(results[0].is_trading, Some(expected_is_trading));
    }

    #[rstest]
    fn test_parse_instrument_statuses_skips_unknown_market_status() {
        // An unmodeled market status must not publish a fabricated non-trading
        // state; the whole definition is skipped.
        let def = make_status_def(MarketStatus::Unknown, true, RunnerStatus::Active);
        let results =
            parse_instrument_statuses("1.123", &def, UnixNanos::default(), UnixNanos::default());

        assert!(results.is_empty());
    }

    #[rstest]
    fn test_parse_instrument_statuses_skips_unknown_runner() {
        // An unmodeled runner status must not be emitted as tradable, even in an
        // open in-play market; we skip it rather than fabricate tradability.
        let def = make_status_def(MarketStatus::Open, true, RunnerStatus::Unknown);
        let results =
            parse_instrument_statuses("1.123", &def, UnixNanos::default(), UnixNanos::default());

        assert!(results.is_empty());
    }

    #[rstest]
    #[case(RunnerStatus::Removed)]
    #[case(RunnerStatus::RemovedVacant)]
    fn test_parse_instrument_statuses_scratched_runner_closes(#[case] runner_status: RunnerStatus) {
        // Even with Open + in_play the runner must close when scratched
        let def = make_status_def(MarketStatus::Open, true, runner_status);
        let results =
            parse_instrument_statuses("1.123", &def, UnixNanos::default(), UnixNanos::default());

        assert_eq!(results.len(), 1);
        assert_eq!(results[0].action, MarketStatusAction::Close);
        assert_eq!(results[0].is_trading, Some(false));
    }

    #[rstest]
    #[case::missing_runners("runners")]
    #[case::missing_status("status")]
    fn test_parse_instrument_statuses_returns_empty(#[case] drop_field: &str) {
        let mut def = make_status_def(MarketStatus::Open, true, RunnerStatus::Active);
        match drop_field {
            "runners" => def.runners = None,
            "status" => def.status = None,
            _ => unreachable!(),
        }

        let results =
            parse_instrument_statuses("1.123", &def, UnixNanos::default(), UnixNanos::default());

        assert!(results.is_empty());
    }

    #[rstest]
    fn test_parse_instrument_statuses_mixed_runners() {
        // Three runners in one market definition: an Active runner should follow the
        // market-level mapping while Removed/RemovedVacant override to Close. Verify
        // that selection id and handicap propagate into the emitted instrument id.
        let mut def = make_status_def(MarketStatus::Open, true, RunnerStatus::Active);
        def.runners = Some(vec![
            RunnerDefinition {
                id: 101,
                hc: None,
                sort_priority: Some(1),
                name: None,
                status: Some(RunnerStatus::Active),
                adjustment_factor: None,
                bsp: None,
                removal_date: None,
            },
            RunnerDefinition {
                id: 202,
                hc: Some(Decimal::new(25, 1)), // 2.5 handicap
                sort_priority: Some(2),
                name: None,
                status: Some(RunnerStatus::Removed),
                adjustment_factor: None,
                bsp: None,
                removal_date: None,
            },
            RunnerDefinition {
                id: 303,
                hc: None,
                sort_priority: Some(3),
                name: None,
                status: Some(RunnerStatus::RemovedVacant),
                adjustment_factor: None,
                bsp: None,
                removal_date: None,
            },
        ]);

        let results =
            parse_instrument_statuses("1.999", &def, UnixNanos::default(), UnixNanos::default());

        assert_eq!(results.len(), 3);

        assert_eq!(results[0].action, MarketStatusAction::Trading);
        assert_eq!(results[0].is_trading, Some(true));

        assert_eq!(results[1].action, MarketStatusAction::Close);
        assert_eq!(results[1].is_trading, Some(false));

        assert_eq!(results[2].action, MarketStatusAction::Close);
        assert_eq!(results[2].is_trading, Some(false));

        // Each runner produces a distinct instrument id (selection id + handicap)
        assert_ne!(results[0].instrument_id, results[1].instrument_id);
        assert_ne!(results[1].instrument_id, results[2].instrument_id);
        assert_ne!(results[0].instrument_id, results[2].instrument_id);
    }

    #[rstest]
    fn test_parse_order_status_report_new_order() {
        let data = load_test_json("stream/ocm_NEW_FULL_IMAGE.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];

            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                parse_millis_timestamp(ocm.pt),
                parse_millis_timestamp(ocm.pt),
            )
            .unwrap();

            // Partially filled: sm=4.75, sr=0.25, status=E
            assert_eq!(report.order_status, OrderStatus::PartiallyFilled);
            assert_eq!(report.order_side, Some(OrderSide::Sell)); // Back → Sell
            assert_eq!(report.order_type, OrderType::Limit);
            assert_eq!(report.filled_qty.as_f64(), 4.75);
            assert_eq!(report.quantity.as_f64(), 5.0);
            assert!(report.price.is_some());
            assert_eq!(report.price.unwrap().as_f64(), 12.0);
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_parse_order_status_report_filled() {
        let data = load_test_json("stream/ocm_FILLED.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];

            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                parse_millis_timestamp(ocm.pt),
                parse_millis_timestamp(ocm.pt),
            )
            .unwrap();

            assert_eq!(report.order_status, OrderStatus::Filled);
            assert_eq!(report.order_side, Some(OrderSide::Buy)); // Lay → Buy
            assert_eq!(report.filled_qty.as_f64(), 10.0);
            assert_eq!(report.quantity.as_f64(), 10.0);

            // Has client_order_id from rfo field
            assert!(report.client_order_id.is_some());
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_parse_order_status_report_cancelled() {
        let data = load_test_json("stream/ocm_CANCEL.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];

            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                parse_millis_timestamp(ocm.pt),
                parse_millis_timestamp(ocm.pt),
            )
            .unwrap();

            assert_eq!(report.order_status, OrderStatus::Canceled);
            assert_eq!(report.order_side, Some(OrderSide::Sell)); // Back → Sell
            assert_eq!(report.filled_qty.as_f64(), 0.0);
            assert_eq!(report.quantity.as_f64(), 10.0);
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_parse_order_status_report_duplicate_execution() {
        let data = load_test_json("stream/ocm_DUPLICATE_EXECUTION.json");
        let msgs: Vec<StreamMessage> = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = &msgs[0] {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];

            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                parse_millis_timestamp(ocm.pt),
                parse_millis_timestamp(ocm.pt),
            )
            .unwrap();

            // Partially filled: sm=1.12, status=E
            assert_eq!(report.order_status, OrderStatus::PartiallyFilled);
            assert_eq!(report.order_side, Some(OrderSide::Buy)); // Lay → Buy
            assert_eq!(report.filled_qty.as_f64(), 1.12);
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_parse_order_status_report_sp_resting() {
        let data = load_test_json("stream/ocm_SP_RESTING.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];

            assert!(is_resting_sp_bet(uo));

            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                parse_millis_timestamp(ocm.pt),
                parse_millis_timestamp(ocm.pt),
            )
            .unwrap();

            // A resting SP bet reports execution-complete with zero size
            // fields, but cannot be cancelled and is still open until BSP
            // reconciliation
            assert_eq!(report.order_status, OrderStatus::Accepted);
            assert_eq!(report.time_in_force, TimeInForce::AtTheClose);
            assert_eq!(report.order_side, Some(OrderSide::Sell)); // Back → Sell
            assert_eq!(report.filled_qty.as_f64(), 0.0);
            assert_eq!(report.quantity.as_f64(), 2.0);
        } else {
            panic!("expected OrderChange");
        }
    }

    fn sp_unmatched_order(size_matched: Decimal, size_lapsed: Decimal) -> UnmatchedOrder {
        UnmatchedOrder {
            id: "442849719274".to_string(),
            p: Decimal::new(10000, 1),
            s: Decimal::ZERO,
            side: StreamingSide::Back,
            status: StreamingOrderStatus::ExecutionComplete,
            pt: Some(StreamingPersistenceType::MarketOnClose),
            ot: StreamingOrderType::MarketOnClose,
            pd: 1789423573000,
            bsp: Some(Decimal::new(2, 0)),
            rfo: None,
            rfs: None,
            rc: None,
            rac: None,
            md: None,
            cd: None,
            ld: None,
            avp: None,
            sm: Some(size_matched),
            sr: Some(Decimal::ZERO),
            sl: Some(size_lapsed),
            sc: Some(Decimal::ZERO),
            sv: Some(Decimal::ZERO),
            lsrc: None,
        }
    }

    #[rstest]
    fn test_parse_order_status_report_sp_reconciled() {
        // At BSP reconciliation an SP bet is no longer resting: a matched bet
        // resolves Filled and a lapsed bet resolves Canceled
        let matched = sp_unmatched_order(Decimal::new(2, 0), Decimal::ZERO);
        assert!(!is_resting_sp_bet(&matched));

        let report = parse_order_status_report(
            &matched,
            InstrumentId::from("1.262362241-6532924-0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.order_status, OrderStatus::Filled);
        assert_eq!(report.filled_qty.as_f64(), 2.0);
        assert_eq!(report.quantity.as_f64(), 2.0);

        let lapsed = sp_unmatched_order(Decimal::ZERO, Decimal::new(2, 0));
        assert!(!is_resting_sp_bet(&lapsed));

        let report = parse_order_status_report(
            &lapsed,
            InstrumentId::from("1.262362241-6532924-0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.order_status, OrderStatus::Canceled);
    }

    #[rstest]
    fn test_parse_order_status_report_multiple_fills() {
        let data = load_test_json("stream/ocm_multiple_fills.json");
        let msgs: Vec<StreamMessage> = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = &msgs[0] {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];

            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                parse_millis_timestamp(ocm.pt),
                parse_millis_timestamp(ocm.pt),
            )
            .unwrap();

            // Partially filled: sm=16.19, status=E, has rfo
            assert_eq!(report.order_status, OrderStatus::PartiallyFilled);
            assert_eq!(report.order_side, Some(OrderSide::Sell)); // Back → Sell
            assert_eq!(report.filled_qty.as_f64(), 16.19);
            assert!(report.client_order_id.is_some());
            assert!(report.avg_px.is_some());
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_parse_runner_book_snapshot_empty_book() {
        // A snapshot with no levels should still produce a clear delta
        let rc = RunnerChange {
            id: 12345,
            hc: None,
            atb: Some(vec![]),
            atl: Some(vec![]),
            batb: None,
            batl: None,
            bdatb: None,
            bdatl: None,
            spb: None,
            spl: None,
            spn: None,
            spf: None,
            trd: None,
            ltp: None,
            tv: None,
        };

        let result = parse_runner_book_deltas(
            make_instrument_id("1.123", 12345, Decimal::ZERO),
            &rc,
            true,
            1000,
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap()
        .expect("should produce snapshot deltas");

        // Just the Clear delta with F_LAST
        assert_eq!(result.deltas.len(), 1);
        assert_eq!(result.deltas[0].action, BookAction::Clear);
        assert!(RecordFlag::F_LAST.matches(result.deltas[0].flags));
        assert!(RecordFlag::F_SNAPSHOT.matches(result.deltas[0].flags));
    }

    #[rstest]
    fn test_make_fill_report() {
        let instrument_id = make_instrument_id("1.180604981", 1209555, Decimal::ZERO);
        let fill = make_fill_report(
            AccountId::from("BETFAIR-001"),
            instrument_id,
            VenueOrderId::from("229430281339"),
            TradeId::from("229430281339-0"),
            OrderSide::Buy,
            Quantity::new(10.0, BETFAIR_QUANTITY_PRECISION),
            Price::new(1.1, BETFAIR_PRICE_PRECISION),
            Currency::GBP(),
            None,
            UnixNanos::default(),
            UnixNanos::default(),
        );

        assert_eq!(fill.instrument_id, instrument_id);
        assert_eq!(fill.order_side, OrderSide::Buy);
        assert_eq!(fill.last_qty.as_f64(), 10.0);
        assert_eq!(fill.last_px.as_f64(), 1.1);
        assert_eq!(fill.commission.as_f64(), 0.0);
        assert_eq!(fill.liquidity_side, LiquiditySide::NoLiquiditySide);
    }

    #[rstest]
    fn test_fill_tracker_single_full_fill() {
        let data = load_test_json("stream/ocm_FILLED.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let fill = tracker
                .maybe_fill_report(
                    uo,
                    uo.s,
                    instrument_id,
                    AccountId::from("BETFAIR-001"),
                    Currency::GBP(),
                    ts,
                    ts,
                )
                .expect("should produce fill");

            assert_eq!(fill.last_qty.as_f64(), 10.0);
            assert_eq!(fill.last_px.as_f64(), 1.1);
            assert_eq!(fill.order_side, OrderSide::Buy);
            assert!(fill.client_order_id.is_some());
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_incremental_fills() {
        let data = load_test_json("stream/ocm_multiple_fills.json");
        let msgs: Vec<StreamMessage> = serde_json::from_str(&data).unwrap();
        let instrument_id = make_instrument_id("1.179082386", 50210, Decimal::ZERO);

        let mut tracker = FillTracker::new();
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::GBP();

        // First fill: sm=16.19 (from zero)
        let uo1 = extract_uo(&msgs[0]);
        let ts1 = extract_ts(&msgs[0]);
        let fill1 = tracker
            .maybe_fill_report(uo1, uo1.s, instrument_id, account_id, currency, ts1, ts1)
            .expect("should produce first fill");
        assert_eq!(fill1.last_qty.as_f64(), 16.19);
        assert_eq!(fill1.last_px.as_f64(), 5.8);

        // Second fill: sm=16.96 (delta=0.77)
        let uo2 = extract_uo(&msgs[1]);
        let ts2 = extract_ts(&msgs[1]);
        let fill2 = tracker
            .maybe_fill_report(uo2, uo2.s, instrument_id, account_id, currency, ts2, ts2)
            .expect("should produce second fill");
        assert_eq!(fill2.last_qty.as_f64(), 0.77);
        assert_eq!(fill2.last_px.as_f64(), 5.8);

        // Third fill: sm=17.73 (delta=0.77)
        let uo3 = extract_uo(&msgs[2]);
        let ts3 = extract_ts(&msgs[2]);
        let fill3 = tracker
            .maybe_fill_report(uo3, uo3.s, instrument_id, account_id, currency, ts3, ts3)
            .expect("should produce third fill");
        assert_eq!(fill3.last_qty.as_f64(), 0.77);
    }

    #[rstest]
    fn test_fill_tracker_different_price() {
        let data = load_test_json("stream/ocm_filled_different_price.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let fill = tracker
                .maybe_fill_report(
                    uo,
                    uo.s,
                    instrument_id,
                    AccountId::from("BETFAIR-001"),
                    Currency::GBP(),
                    ts,
                    ts,
                )
                .expect("should produce fill");

            // Order placed at 1.3 but average fill price is 1.2
            assert_eq!(fill.last_qty.as_f64(), 20.0);
            assert_eq!(fill.last_px.as_f64(), 1.2);
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_cancel_no_fill() {
        let data = load_test_json("stream/ocm_CANCEL.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let result = tracker.maybe_fill_report(
                uo,
                uo.s,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                Currency::GBP(),
                ts,
                ts,
            );
            assert!(result.is_none(), "cancelled order should not produce fill");
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_lapsed_no_fill() {
        let data = load_test_json("stream/ocm_error_fill.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let result = tracker.maybe_fill_report(
                uo,
                uo.s,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                Currency::GBP(),
                ts,
                ts,
            );
            assert!(result.is_none(), "lapsed order should not produce fill");
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_duplicate_dedup() {
        let data = load_test_json("stream/ocm_FILLED.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);
            let account_id = AccountId::from("BETFAIR-001");
            let currency = Currency::GBP();

            let mut tracker = FillTracker::new();

            // First call produces fill
            let fill1 =
                tracker.maybe_fill_report(uo, uo.s, instrument_id, account_id, currency, ts, ts);
            assert!(fill1.is_some());

            // Second call with same data produces nothing (dedup)
            let fill2 =
                tracker.maybe_fill_report(uo, uo.s, instrument_id, account_id, currency, ts, ts);
            assert!(fill2.is_none(), "duplicate fill should be suppressed");
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_price_back_calculation() {
        let data = load_test_json("stream/ocm_multiple_fills.json");
        let msgs: Vec<StreamMessage> = serde_json::from_str(&data).unwrap();
        let instrument_id = make_instrument_id("1.179082386", 50210, Decimal::ZERO);
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::GBP();
        let mut tracker = FillTracker::new();

        // Process first fill at avp=5.8
        let uo1 = extract_uo(&msgs[0]);
        let ts1 = extract_ts(&msgs[0]);
        let fill1 = tracker
            .maybe_fill_report(uo1, uo1.s, instrument_id, account_id, currency, ts1, ts1)
            .unwrap();
        assert_eq!(fill1.last_px.as_f64(), 5.8);

        // Second fill also at avp=5.8 (same price, avg unchanged)
        let uo2 = extract_uo(&msgs[1]);
        let ts2 = extract_ts(&msgs[1]);
        let fill2 = tracker
            .maybe_fill_report(uo2, uo2.s, instrument_id, account_id, currency, ts2, ts2)
            .unwrap();
        assert_eq!(fill2.last_px.as_f64(), 5.8);
    }

    #[rstest]
    fn test_has_cancel_quantity() {
        let data = load_test_json("stream/ocm_CANCEL.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let uo = &ocm.oc.as_ref().unwrap()[0].orc.as_ref().unwrap()[0]
                .uo
                .as_ref()
                .unwrap()[0];
            assert!(has_cancel_quantity(uo));
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_has_cancel_quantity_filled_order() {
        let data = load_test_json("stream/ocm_FILLED.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let uo = &ocm.oc.as_ref().unwrap()[0].orc.as_ref().unwrap()[0]
                .uo
                .as_ref()
                .unwrap()[0];
            assert!(!has_cancel_quantity(uo));
        } else {
            panic!("expected OrderChange");
        }
    }

    fn extract_uo(msg: &StreamMessage) -> &UnmatchedOrder {
        if let StreamMessage::OrderChange(ocm) = msg {
            &ocm.oc.as_ref().unwrap()[0].orc.as_ref().unwrap()[0]
                .uo
                .as_ref()
                .unwrap()[0]
        } else {
            panic!("expected OrderChange")
        }
    }

    fn extract_ts(msg: &StreamMessage) -> UnixNanos {
        if let StreamMessage::OrderChange(ocm) = msg {
            parse_millis_timestamp(ocm.pt)
        } else {
            panic!("expected OrderChange")
        }
    }

    #[rstest]
    fn test_parse_race_runner_data_from_fixture() {
        let data = load_test_json("stream/rcm_single.json");
        let msg = stream_decode(data.as_bytes()).unwrap();

        let StreamMessage::RaceChange(rcm) = msg else {
            panic!("expected RaceChange");
        };

        let race = &rcm.rc.as_ref().unwrap()[0];
        let rrc = &race.rrc.as_ref().unwrap()[0];
        let ts = parse_millis_timestamp(rcm.pt);

        let runner = parse_race_runner_data(
            race.id.as_deref().unwrap(),
            race.mid.as_deref().unwrap(),
            rrc,
            ts,
            ts,
        )
        .unwrap();

        assert_eq!(runner.race_id, "28587288.1650");
        assert_eq!(runner.market_id, "1.1234567");
        assert_eq!(runner.selection_id, 7390417);
        assert!((runner.latitude - 51.4189543).abs() < 1e-6);
        assert!((runner.longitude - (-0.4058491)).abs() < 1e-6);
        assert!((runner.speed - 17.8).abs() < 1e-6);
        assert!((runner.progress - 2051.0).abs() < 1e-6);
        assert!((runner.stride_frequency - 2.07).abs() < 1e-6);
    }

    #[rstest]
    fn test_parse_race_progress_from_fixture() {
        let data = load_test_json("stream/rcm_single.json");
        let msg = stream_decode(data.as_bytes()).unwrap();

        let StreamMessage::RaceChange(rcm) = msg else {
            panic!("expected RaceChange");
        };

        let race = &rcm.rc.as_ref().unwrap()[0];
        let rpc = race.rpc.as_ref().unwrap();
        let ts = parse_millis_timestamp(rcm.pt);

        let progress = parse_race_progress(
            race.id.as_deref().unwrap(),
            race.mid.as_deref().unwrap(),
            rpc,
            ts,
            ts,
        );

        assert_eq!(progress.race_id, "28587288.1650");
        assert_eq!(progress.market_id, "1.1234567");
        assert_eq!(progress.gate_name, "1f");
        assert!((progress.sectional_time - 10.6).abs() < 1e-6);
        assert!((progress.running_time - 46.7).abs() < 1e-6);
        assert!((progress.speed - 17.8).abs() < 1e-6);
        assert!((progress.progress - 87.5).abs() < 1e-6);

        let order: Vec<i64> = serde_json::from_str(&progress.order).unwrap();
        assert_eq!(order, vec![7390417, 5600338, 11527189, 6395118, 8706072]);

        let jumps: Vec<serde_json::Value> = serde_json::from_str(&progress.jumps).unwrap();
        assert_eq!(jumps.len(), 2);
        assert_eq!(jumps[0]["J"], 2);
    }

    #[rstest]
    fn test_parse_cricket_match_from_fixture() {
        let data = load_test_json("stream/ccm_single.json");
        let msg = stream_decode(data.as_bytes()).unwrap();

        let StreamMessage::CricketChange(ccm) = msg else {
            panic!("expected CricketChange");
        };

        let change = &ccm.cc.as_ref().unwrap()[0];
        let ts = parse_millis_timestamp(ccm.pt);
        let cricket = parse_cricket_match(change, ts, ts).unwrap();

        assert_eq!(cricket.event_id, "35741575");
        assert_eq!(cricket.market_id, "1.259334639");
        let stats: serde_json::Value = serde_json::from_str(&cricket.match_stats).unwrap();
        assert_eq!(stats["inningsStats"][0]["inningsRuns"], 100);
    }

    #[rstest]
    #[case(None, Some("1.259334639".to_string()))]
    #[case(Some("35741575".to_string()), None)]
    fn test_parse_cricket_match_missing_required_id_returns_none(
        #[case] event_id: Option<String>,
        #[case] market_id: Option<String>,
    ) {
        let change = CricketChange {
            event_id,
            market_id,
            fixture_info: None,
            home_team: None,
            away_team: None,
            match_stats: Some(serde_json::json!({"inningsStats": []})),
            incident_list_wrapper: None,
        };
        let ts = UnixNanos::from(1_000_000_000u64);
        let result = parse_cricket_match(&change, ts, ts);

        assert!(result.is_none());
    }

    #[rstest]
    fn test_parse_race_runner_data_multi_runner() {
        let data = load_test_json("stream/rcm_multi_runner.json");
        let msg = stream_decode(data.as_bytes()).unwrap();

        let StreamMessage::RaceChange(rcm) = msg else {
            panic!("expected RaceChange");
        };

        let race = &rcm.rc.as_ref().unwrap()[0];
        let ts = parse_millis_timestamp(rcm.pt);
        let race_id = race.id.as_deref().unwrap();
        let market_id = race.mid.as_deref().unwrap();

        let runners: Vec<_> = race
            .rrc
            .as_ref()
            .unwrap()
            .iter()
            .filter_map(|rrc| parse_race_runner_data(race_id, market_id, rrc, ts, ts))
            .collect();

        assert_eq!(runners.len(), 5);
        assert_eq!(runners[0].selection_id, 35467839);
        assert_eq!(runners[4].selection_id, 41694785);
        assert!((runners[0].speed - 16.33).abs() < 1e-6);
        assert!((runners[4].speed - 17.11).abs() < 1e-6);
    }

    #[rstest]
    fn test_parse_race_runner_data_missing_id_returns_none() {
        let rrc = RaceRunnerChange {
            ft: Some(1000),
            id: None,
            lat: Some(51.0),
            lng: Some(-0.4),
            spd: Some(15.0),
            prg: Some(500.0),
            sfq: Some(2.0),
        };
        let ts = UnixNanos::from(1_000_000_000u64);
        let result = parse_race_runner_data("race1", "market1", &rrc, ts, ts);
        assert!(result.is_none());
    }

    #[rstest]
    fn test_parse_race_runner_data_absent_fields_are_nan() {
        let rrc = RaceRunnerChange {
            ft: None,
            id: Some(12345),
            lat: None,
            lng: None,
            spd: None,
            prg: None,
            sfq: None,
        };
        let ts = UnixNanos::from(1_000_000_000u64);
        let runner = parse_race_runner_data("race1", "market1", &rrc, ts, ts).unwrap();
        assert!(runner.latitude.is_nan());
        assert!(runner.longitude.is_nan());
        assert!(runner.speed.is_nan());
        assert!(runner.progress.is_nan());
        assert!(runner.stride_frequency.is_nan());
    }

    #[rstest]
    fn test_parse_race_progress_absent_fields() {
        let rpc = RaceProgressChange {
            ft: None,
            g: None,
            st: None,
            rt: None,
            spd: None,
            prg: None,
            ord: None,
            jumps: None,
        };
        let ts = UnixNanos::from(1_000_000_000u64);
        let progress = parse_race_progress("race1", "market1", &rpc, ts, ts);
        assert_eq!(progress.gate_name, "");
        assert!(progress.sectional_time.is_nan());
        assert!(progress.running_time.is_nan());
        assert_eq!(progress.order, "");
        assert_eq!(progress.jumps, "");
    }

    fn runner_change_with_ticker(
        id: u64,
        ltp: Option<Decimal>,
        tv: Option<Decimal>,
        spn: Option<Decimal>,
        spf: Option<Decimal>,
    ) -> RunnerChange {
        RunnerChange {
            id,
            hc: None,
            atb: None,
            atl: None,
            batb: None,
            batl: None,
            bdatb: None,
            bdatl: None,
            spb: None,
            spl: None,
            spn,
            spf,
            trd: None,
            ltp,
            tv,
        }
    }

    #[rstest]
    fn test_parse_betfair_ticker_all_fields() {
        let rc = runner_change_with_ticker(
            9249757,
            Some(Decimal::new(55, 1)),
            Some(Decimal::new(189032, 2)),
            Some(Decimal::new(568, 2)),
            Some(Decimal::new(573, 2)),
        );
        let ts = UnixNanos::from(1_000_000_000u64);
        let instrument_id = make_instrument_id("1.185781465", 9249757, Decimal::ZERO);

        let ticker = parse_betfair_ticker(instrument_id, &rc, ts, ts).unwrap();

        assert_eq!(ticker.instrument_id, instrument_id);
        assert_decimal_option_eq(ticker.last_traded_price, Decimal::new(55, 1));
        assert_decimal_option_eq(ticker.traded_volume, Decimal::new(189032, 2));
        assert_decimal_option_eq(ticker.starting_price_near, Decimal::new(568, 2));
        assert_decimal_option_eq(ticker.starting_price_far, Decimal::new(573, 2));
    }

    #[rstest]
    fn test_parse_betfair_ticker_partial_fields() {
        let rc = runner_change_with_ticker(
            9249757,
            Some(Decimal::new(55, 1)),
            Some(Decimal::new(189032, 2)),
            None,
            None,
        );
        let ts = UnixNanos::from(1_000_000_000u64);
        let instrument_id = make_instrument_id("1.185781465", 9249757, Decimal::ZERO);

        let ticker = parse_betfair_ticker(instrument_id, &rc, ts, ts).unwrap();

        assert_decimal_option_eq(ticker.last_traded_price, Decimal::new(55, 1));
        assert_decimal_option_eq(ticker.traded_volume, Decimal::new(189032, 2));
        assert!(ticker.starting_price_near.is_none());
        assert!(ticker.starting_price_far.is_none());
    }

    #[rstest]
    fn test_parse_betfair_ticker_no_fields_returns_none() {
        let rc = runner_change_with_ticker(9249757, None, None, None, None);
        let ts = UnixNanos::from(1_000_000_000u64);
        let instrument_id = make_instrument_id("1.185781465", 9249757, Decimal::ZERO);

        assert!(parse_betfair_ticker(instrument_id, &rc, ts, ts).is_none());
    }

    #[rstest]
    fn test_parse_betfair_ticker_only_tv() {
        let rc =
            runner_change_with_ticker(40273293, None, Some(Decimal::new(320115, 2)), None, None);
        let ts = UnixNanos::from(1_000_000_000u64);
        let instrument_id = make_instrument_id("1.185781465", 40273293, Decimal::ZERO);

        let ticker = parse_betfair_ticker(instrument_id, &rc, ts, ts).unwrap();

        assert!(ticker.last_traded_price.is_none());
        assert_decimal_option_eq(ticker.traded_volume, Decimal::new(320115, 2));
        assert!(ticker.starting_price_near.is_none());
        assert!(ticker.starting_price_far.is_none());
    }

    #[rstest]
    fn test_parse_betfair_ticker_from_fixture() {
        let data = load_test_json("stream/mcm_BSP_settled.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::MarketChange(mcm) = msg {
            let mc = mcm.mc.as_ref().unwrap();
            let change = &mc[0];
            let rc_list = change.rc.as_ref().unwrap();

            // Runner 9249757 has ltp=5.5, tv=1890.32, spn=5.68, spf=5.73
            let rc = rc_list.iter().find(|r| r.id == 9249757).unwrap();
            let instrument_id = make_instrument_id(&change.id, rc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(mcm.pt);

            let ticker = parse_betfair_ticker(instrument_id, rc, ts, ts).unwrap();

            assert_decimal_option_eq(ticker.last_traded_price, Decimal::new(55, 1));
            assert_decimal_option_eq(ticker.traded_volume, Decimal::new(189032, 2));
            assert_decimal_option_eq(ticker.starting_price_near, Decimal::new(568, 2));
            assert_decimal_option_eq(ticker.starting_price_far, Decimal::new(573, 2));

            // Runner 40273293 has ltp=2.1, tv=3201.15 but no spn/spf
            let rc2 = rc_list.iter().find(|r| r.id == 40273293).unwrap();
            let instrument_id2 = make_instrument_id(&change.id, rc2.id, Decimal::ZERO);
            let ticker2 = parse_betfair_ticker(instrument_id2, rc2, ts, ts).unwrap();

            assert_decimal_option_eq(ticker2.last_traded_price, Decimal::new(21, 1));
            assert_decimal_option_eq(ticker2.traded_volume, Decimal::new(320115, 2));
            assert!(ticker2.starting_price_near.is_none());
            assert!(ticker2.starting_price_far.is_none());

            // Runner 23678734 has only tv=0, no ltp
            let rc3 = rc_list.iter().find(|r| r.id == 23678734).unwrap();
            let instrument_id3 = make_instrument_id(&change.id, rc3.id, Decimal::ZERO);
            let ticker3 = parse_betfair_ticker(instrument_id3, rc3, ts, ts).unwrap();

            assert!(ticker3.last_traded_price.is_none());
            assert_decimal_option_eq(ticker3.traded_volume, Decimal::ZERO);
        } else {
            panic!("Expected MarketChange");
        }
    }

    #[rstest]
    fn test_parse_betfair_starting_prices_from_fixture() {
        let data = load_test_json("stream/mcm_BSP_settled.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::MarketChange(mcm) = msg {
            let mc = mcm.mc.as_ref().unwrap();
            let def = mc[0].market_definition.as_ref().unwrap();
            let ts = parse_millis_timestamp(mcm.pt);

            let prices = parse_betfair_starting_prices(&mc[0].id, def, ts, ts);

            // 3 runners have bsp values, 1 (REMOVED) does not
            assert_eq!(prices.len(), 3);

            let bsp_map: std::collections::HashMap<String, Decimal> = prices
                .iter()
                .map(|p| (p.instrument_id.to_string(), p.bsp))
                .collect();

            let id_winner = make_instrument_id("1.185781465", 9249757, Decimal::ZERO).to_string();
            let id_placed = make_instrument_id("1.185781465", 40273293, Decimal::ZERO).to_string();
            let id_loser = make_instrument_id("1.185781465", 11120000, Decimal::ZERO).to_string();

            assert_eq!(bsp_map[&id_winner], Decimal::new(573, 2));
            assert_eq!(bsp_map[&id_placed], Decimal::new(214, 2));
            assert_eq!(bsp_map[&id_loser], Decimal::new(2856, 2));
        } else {
            panic!("Expected MarketChange");
        }
    }

    #[rstest]
    fn test_parse_betfair_starting_prices_no_runners() {
        let def = MarketDefinition {
            runners: None,
            bet_delay: None,
            betting_type: None,
            bsp_market: None,
            bsp_reconciled: None,
            competition_id: None,
            competition_name: None,
            complete: None,
            country_code: None,
            cross_matching: None,
            discount_allowed: None,
            each_way_divisor: None,
            event_id: None,
            event_name: None,
            event_type_id: None,
            event_type_name: None,
            in_play: None,
            line_interval: None,
            line_max_unit: None,
            line_min_unit: None,
            market_base_rate: None,
            market_id: None,
            market_name: None,
            market_time: None,
            market_type: None,
            number_of_active_runners: None,
            number_of_winners: None,
            open_date: None,
            persistence_enabled: None,
            price_ladder_definition: None,
            race_type: None,
            regulators: None,
            runners_voidable: None,
            settled_time: None,
            status: None,
            suspend_time: None,
            timezone: None,
            turn_in_play_enabled: None,
            venue: None,
            version: None,
        };
        let ts = UnixNanos::from(1_000_000_000u64);

        let prices = parse_betfair_starting_prices("1.12345", &def, ts, ts);

        assert!(prices.is_empty());
    }

    #[rstest]
    fn test_parse_bsp_book_deltas_from_fixture() {
        let data = load_test_json("stream/mcm_BSP.json");
        let messages: Vec<StreamMessage> = serde_json::from_str(&data).unwrap();

        // Find the MCM with runner changes containing spb/spl data
        let mcm = messages
            .iter()
            .find_map(|m| match m {
                StreamMessage::MarketChange(mcm) => {
                    let mc = mcm.mc.as_ref()?;
                    let has_spb = mc.iter().any(|c| {
                        c.rc.as_ref()
                            .is_some_and(|rcs| rcs.iter().any(|r| r.spb.is_some()))
                    });

                    if has_spb { Some(mcm) } else { None }
                }
                _ => None,
            })
            .expect("fixture should contain MCM with spb data");

        let mc = mcm.mc.as_ref().unwrap();
        let change = &mc[0];
        let rc_list = change.rc.as_ref().unwrap();

        // Runner 9249757 has spb and spl arrays
        let rc = rc_list.iter().find(|r| r.id == 9249757).unwrap();
        let instrument_id = make_instrument_id(&change.id, rc.id, Decimal::ZERO);
        let ts = parse_millis_timestamp(mcm.pt);

        let deltas = parse_bsp_book_deltas(instrument_id, rc, ts, ts);

        let spb_count = rc.spb.as_ref().unwrap().len();
        let spl_count = rc.spl.as_ref().unwrap().len();
        assert_eq!(deltas.len(), spb_count + spl_count);

        // SPB entries are Sell side
        assert_eq!(deltas[0].side, OrderSide::Sell);
        assert_eq!(deltas[0].price, Decimal::new(1000, 0));
        assert_eq!(deltas[0].size, Decimal::new(3338, 2));
        assert_eq!(deltas[0].action, BookAction::Update);

        // SPL entries are Buy side
        let spl_start = spb_count;
        assert_eq!(deltas[spl_start].side, OrderSide::Buy);
        assert_eq!(deltas[spl_start].price, Decimal::new(7, 0));
        assert_eq!(deltas[spl_start].size, Decimal::new(10, 0));
    }

    #[rstest]
    fn test_parse_bsp_book_deltas_zero_volume_is_delete() {
        let rc = RunnerChange {
            id: 12345,
            hc: None,
            atb: None,
            atl: None,
            batb: None,
            batl: None,
            bdatb: None,
            bdatl: None,
            spb: Some(vec![PV {
                price: Decimal::new(50, 1),
                volume: Decimal::ZERO,
            }]),
            spl: None,
            spn: None,
            spf: None,
            trd: None,
            ltp: None,
            tv: None,
        };
        let ts = UnixNanos::from(1_000_000_000u64);
        let instrument_id = make_instrument_id("1.12345", 12345, Decimal::ZERO);

        let deltas = parse_bsp_book_deltas(instrument_id, &rc, ts, ts);

        assert_eq!(deltas.len(), 1);
        assert_eq!(deltas[0].action, BookAction::Delete);
        assert_eq!(deltas[0].price, Decimal::new(5, 0));
        assert_eq!(deltas[0].size, Decimal::ZERO);
    }

    #[rstest]
    fn test_parse_bsp_book_deltas_no_spb_spl_returns_empty() {
        let rc = runner_change_with_ticker(12345, None, None, None, None);
        let ts = UnixNanos::from(1_000_000_000u64);
        let instrument_id = make_instrument_id("1.12345", 12345, Decimal::ZERO);

        let deltas = parse_bsp_book_deltas(instrument_id, &rc, ts, ts);

        assert!(deltas.is_empty());
    }

    #[rstest]
    fn test_parse_instrument_closes_from_fixture() {
        let data = load_test_json("stream/mcm_BSP_settled.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::MarketChange(mcm) = msg {
            let mc = mcm.mc.as_ref().unwrap();
            let def = mc[0].market_definition.as_ref().unwrap();
            let ts = parse_millis_timestamp(mcm.pt);

            let closes = parse_instrument_closes(&mc[0].id, def, ts, ts);

            // 4 runners: WINNER, PLACED, LOSER, REMOVED - all produce close events
            assert_eq!(closes.len(), 4);

            let close_map: std::collections::HashMap<String, Price> = closes
                .iter()
                .map(|c| (c.instrument_id.to_string(), c.close_price))
                .collect();

            let id_winner = make_instrument_id("1.185781465", 9249757, Decimal::ZERO).to_string();
            let id_placed = make_instrument_id("1.185781465", 40273293, Decimal::ZERO).to_string();
            let id_loser = make_instrument_id("1.185781465", 11120000, Decimal::ZERO).to_string();
            let id_removed = make_instrument_id("1.185781465", 37433527, Decimal::ZERO).to_string();

            assert_eq!(close_map[&id_winner], Price::from("1.00"));
            assert_eq!(close_map[&id_placed], Price::from("1.00"));
            assert_eq!(close_map[&id_loser], Price::from("0.00"));
            assert_eq!(close_map[&id_removed], Price::from("0.00"));
        } else {
            panic!("Expected MarketChange");
        }
    }

    #[rstest]
    fn test_parse_instrument_closes_active_runners_excluded() {
        let data = load_test_json("stream/mcm_BSP.json");
        let messages: Vec<StreamMessage> = serde_json::from_str(&data).unwrap();

        // Find the MCM with a market definition containing ACTIVE runners
        let mcm = messages
            .iter()
            .find_map(|m| match m {
                StreamMessage::MarketChange(mcm) => {
                    let mc = mcm.mc.as_ref()?;
                    mc.iter()
                        .find(|c| c.market_definition.is_some())
                        .map(|_| mcm)
                }
                _ => None,
            })
            .expect("fixture should contain MCM with market definition");

        let mc = mcm.mc.as_ref().unwrap();
        let change = mc.iter().find(|c| c.market_definition.is_some()).unwrap();
        let def = change.market_definition.as_ref().unwrap();
        let ts = parse_millis_timestamp(mcm.pt);

        let closes = parse_instrument_closes(&change.id, def, ts, ts);

        assert!(
            closes.is_empty(),
            "Active runners should not produce close events, found {}",
            closes.len()
        );
    }

    fn make_test_uo(
        bet_id: &str,
        size: Decimal,
        sm: Option<Decimal>,
        avp: Option<Decimal>,
    ) -> UnmatchedOrder {
        UnmatchedOrder {
            id: bet_id.to_string(),
            p: Decimal::new(25, 1),
            s: size,
            side: StreamingSide::Back,
            status: StreamingOrderStatus::Executable,
            pt: Some(StreamingPersistenceType::Lapse),
            ot: StreamingOrderType::Limit,
            pd: 1616568581000,
            bsp: None,
            rfo: None,
            rfs: None,
            rc: None,
            rac: None,
            md: None,
            cd: None,
            ld: None,
            avp,
            sm,
            sr: None,
            sl: None,
            sc: None,
            sv: None,
            lsrc: None,
        }
    }

    #[rstest]
    fn test_make_trade_id_normalizes_float_tail_sm() {
        let uo = make_test_uo(
            "430069890490",
            Decimal::new(4287, 2),
            Some(Decimal::new(4_287_000_000_000_001, 14)),
            Some(Decimal::new(25, 1)),
        );

        let trade_id = make_trade_id(&uo);

        assert_eq!(trade_id.as_str(), "430069890490-42.87");
    }

    #[rstest]
    fn test_fill_tracker_sync_order_prevents_duplicate_fill() {
        let mut tracker = FillTracker::new();

        // Sync existing fill state: 10 matched at 2.5
        tracker.sync_order("123456", Decimal::new(10, 0), Decimal::new(25, 1));

        let uo = make_test_uo(
            "123456",
            Decimal::new(20, 0),
            Some(Decimal::new(10, 0)),
            Some(Decimal::new(25, 1)),
        );

        let instrument_id = InstrumentId::from("1.234567-123456-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::from("GBP");
        let ts = UnixNanos::default();

        // Same sm=10 as synced, should not emit a fill
        let result =
            tracker.maybe_fill_report(&uo, uo.s, instrument_id, account_id, currency, ts, ts);
        assert!(
            result.is_none(),
            "should not emit fill for already-synced qty"
        );
    }

    #[rstest]
    fn test_fill_tracker_sync_order_is_monotonic() {
        let mut tracker = FillTracker::new();

        // Tracker already has 10 from prior stream activity
        tracker.sync_order("123456", Decimal::new(10, 0), Decimal::new(25, 1));

        // Cache lags (engine hasn't processed the last fill yet) and reports 5
        tracker.sync_order("123456", Decimal::new(5, 0), Decimal::new(20, 1));

        // Next OCM with sm=15 must emit incremental fill of 5, not 10
        let uo = make_test_uo(
            "123456",
            Decimal::new(20, 0),
            Some(Decimal::new(15, 0)),
            Some(Decimal::new(25, 1)),
        );
        let result = tracker.maybe_fill_report(
            &uo,
            uo.s,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            Currency::from("GBP"),
            UnixNanos::default(),
            UnixNanos::default(),
        );
        let fill = result.expect("should emit incremental fill");
        assert_eq!(fill.last_qty, Quantity::from("5.00"));
    }

    #[rstest]
    fn test_fill_tracker_sync_order_allows_incremental_fill() {
        let mut tracker = FillTracker::new();

        // Sync: 10 matched at 2.5
        tracker.sync_order("123456", Decimal::new(10, 0), Decimal::new(25, 1));

        let uo = make_test_uo(
            "123456",
            Decimal::new(20, 0),
            Some(Decimal::new(15, 0)),
            Some(Decimal::new(26, 1)),
        );

        let instrument_id = InstrumentId::from("1.234567-123456-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::from("GBP");
        let ts = UnixNanos::default();

        // sm=15 vs synced 10, should emit incremental fill of 5
        let result =
            tracker.maybe_fill_report(&uo, uo.s, instrument_id, account_id, currency, ts, ts);
        assert!(result.is_some(), "should emit fill for new matched qty");
        let fill = result.unwrap();
        assert_eq!(fill.last_qty, Quantity::from("5.00"));
    }

    #[rstest]
    fn test_fill_tracker_seed_published_trade_ids_blocks_replay() {
        let mut tracker = FillTracker::new();

        let uo_for_id = make_test_uo(
            "123456",
            Decimal::new(20, 0),
            Some(Decimal::new(10, 0)),
            Some(Decimal::new(25, 1)),
        );
        let trade_id = make_trade_id(&uo_for_id);

        // Seeded trade-id must block even with an empty FillTracker
        // (cumulative-size gate would otherwise let this fill through)
        tracker.seed_published_trade_ids([trade_id.to_string()]);

        let result = tracker.maybe_fill_report(
            &uo_for_id,
            uo_for_id.s,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            Currency::from("GBP"),
            UnixNanos::default(),
            UnixNanos::default(),
        );

        assert!(
            result.is_none(),
            "seeded trade-id should suppress duplicate fill",
        );

        let next_update = make_test_uo(
            "123456",
            Decimal::new(20, 0),
            Some(Decimal::new(15, 0)),
            Some(Decimal::new(25, 1)),
        );
        let next_fill = tracker
            .maybe_fill_report(
                &next_update,
                next_update.s,
                InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
                AccountId::from("BETFAIR-001"),
                Currency::from("GBP"),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .expect("later cumulative state should emit only the new delta");
        assert_eq!(next_fill.last_qty, Quantity::from("5.00"));
    }

    #[rstest]
    fn test_fill_tracker_accepts_sm_float_tail_at_order_quantity() {
        let mut tracker = FillTracker::new();
        let uo = make_test_uo(
            "430069890490",
            Decimal::new(4287, 2),
            Some(Decimal::new(4_287_000_000_000_001, 14)),
            Some(Decimal::new(25, 1)),
        );

        let result = tracker.maybe_fill_report(
            &uo,
            uo.s,
            InstrumentId::from("1.234567-430069890490-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            Currency::from("GBP"),
            UnixNanos::default(),
            UnixNanos::default(),
        );

        let fill = result.expect("float tail at order quantity should emit fill");
        assert_eq!(fill.last_qty, Quantity::from("42.87"));
        assert_eq!(fill.trade_id.as_str(), "430069890490-42.87");

        let duplicate = tracker.maybe_fill_report(
            &uo,
            uo.s,
            InstrumentId::from("1.234567-430069890490-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            Currency::from("GBP"),
            UnixNanos::default(),
            UnixNanos::default(),
        );
        assert!(
            duplicate.is_none(),
            "same normalized sm should not emit a duplicate fill"
        );
    }

    #[rstest]
    fn test_fill_tracker_normalized_duplicate_updates_average_price_anchor() {
        let mut tracker = FillTracker::new();

        let first = tracker.advance_cumulative_fill(
            "123456",
            Decimal::new(10, 0),
            Some(Decimal::new(20, 1)),
            Decimal::new(20, 1),
        );
        let normalized_duplicate = tracker.advance_cumulative_fill(
            "123456",
            Decimal::new(10001, 3),
            Some(Decimal::new(30, 1)),
            Decimal::new(30, 1),
        );
        let next = tracker
            .advance_cumulative_fill(
                "123456",
                Decimal::new(11, 0),
                Some(Decimal::new(30, 1)),
                Decimal::new(30, 1),
            )
            .expect("new cumulative quantity should emit a fill");

        let first = first.expect("first fill");
        let restatement = normalized_duplicate.expect("changed average restates the surviving lot");
        assert_eq!(restatement.0, first.0);
        assert_eq!(restatement.1, Quantity::from("10.00"));
        assert_eq!(restatement.2, Price::from("3.00"));
        assert_eq!(next.1, Quantity::from("1.00"));
        assert_eq!(next.2, Price::from("3.00"));
    }

    #[rstest]
    fn test_fill_tracker_changed_average_with_two_lots_emits_nothing() {
        let mut tracker = FillTracker::new();
        let first = tracker
            .advance_cumulative_fill(
                "bet",
                Decimal::from(2),
                Some(Decimal::from(12)),
                Decimal::from(12),
            )
            .expect("first lot");
        let second = tracker
            .advance_cumulative_fill(
                "bet",
                Decimal::from(4),
                Some(Decimal::from(10)),
                Decimal::from(10),
            )
            .expect("second lot");
        let restatement = tracker.advance_cumulative_fill(
            "bet",
            Decimal::from(4),
            Some(Decimal::new(900, 2)),
            Decimal::from(12),
        );
        let later = tracker
            .advance_cumulative_fill(
                "bet",
                Decimal::from(6),
                Some(Decimal::from(8)),
                Decimal::from(12),
            )
            .expect("later fill uses the corrected average");

        assert_ne!(first.0, second.0);
        assert!(restatement.is_none());
        // (8 * 6 - 9 * 4) / 2 = 6, not (8 * 6 - 10 * 4) / 2 = 4
        assert_eq!(later.1, Quantity::from("2.00"));
        assert_eq!(later.2, Price::from("6.00"));
    }

    #[rstest]
    fn test_fill_tracker_void_with_new_average_does_not_restate_price() {
        let mut tracker = FillTracker::new();
        tracker
            .advance_cumulative_fill(
                "bet",
                Decimal::from(2),
                Some(Decimal::from(12)),
                Decimal::from(12),
            )
            .expect("fill");
        let restatement = tracker.advance_cumulative_fill_with_voids(
            "bet",
            Decimal::from(2),
            Decimal::from(1),
            Some(Decimal::new(947, 2)),
            Decimal::from(12),
        );

        assert!(restatement.is_none());
    }

    #[rstest]
    fn test_fill_tracker_post_void_fill_uses_restated_accounting_price() {
        let mut tracker = FillTracker::new();
        let bet_id = "bet-void-restate";
        tracker
            .advance_cumulative_fill(
                bet_id,
                Decimal::from(2),
                Some(Decimal::from(12)),
                Decimal::from(12),
            )
            .expect("fill");

        let voided = tracker.maybe_fill_voids(&UnmatchedOrder {
            id: bet_id.to_string(),
            sv: Some(Decimal::from(1)),
            ..make_test_uo(bet_id, Decimal::from(2), Some(Decimal::from(1)), None)
        });

        let restatement = tracker
            .advance_cumulative_fill_with_voids(
                bet_id,
                Decimal::from(2),
                Decimal::from(1),
                Some(Decimal::new(947, 2)),
                Decimal::from(12),
            )
            .expect("one surviving lot restates");
        let later = tracker
            .advance_cumulative_fill_with_voids(
                bet_id,
                Decimal::from(3),
                Decimal::from(1),
                Some(Decimal::from(10)),
                Decimal::from(12),
            )
            .expect("later fill");

        assert_eq!(voided[0].last_px, Price::from("12.00"));
        assert_eq!(restatement.2, Price::from("9.47"));
        assert_eq!(later.2, Price::from("10.53"));
    }

    #[rstest]
    fn test_fill_tracker_overfill_rejected() {
        let mut tracker = FillTracker::new();

        // sm=30 exceeds order size s=20
        let uo = make_test_uo(
            "999001",
            Decimal::new(20, 0),
            Some(Decimal::new(30, 0)),
            Some(Decimal::new(25, 1)),
        );

        let instrument_id = InstrumentId::from("1.234567-999001-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::from("GBP");
        let ts = UnixNanos::default();

        let result =
            tracker.maybe_fill_report(&uo, uo.s, instrument_id, account_id, currency, ts, ts);
        assert!(
            result.is_none(),
            "overfill (sm > order_qty) should be rejected"
        );
    }

    #[rstest]
    fn test_fill_tracker_zero_sm_returns_none() {
        let mut tracker = FillTracker::new();

        let uo = make_test_uo("999002", Decimal::new(10, 0), Some(Decimal::ZERO), None);

        let instrument_id = InstrumentId::from("1.234567-999002-0.0.BETFAIR");
        let result = tracker.maybe_fill_report(
            &uo,
            uo.s,
            instrument_id,
            AccountId::from("BETFAIR-001"),
            Currency::from("GBP"),
            UnixNanos::default(),
            UnixNanos::default(),
        );
        assert!(result.is_none(), "zero sm should not produce a fill");
    }

    #[rstest]
    fn test_fill_tracker_no_avp_uses_order_price() {
        let data = load_test_json("stream/ocm_FILLED_no_avp.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let fill = tracker
                .maybe_fill_report(
                    uo,
                    uo.s,
                    instrument_id,
                    AccountId::from("BETFAIR-001"),
                    Currency::GBP(),
                    ts,
                    ts,
                )
                .expect("should produce fill even without avp");

            // No avp field, so fill price falls back to order price (p=3.5)
            assert_eq!(fill.last_qty.as_f64(), 25.0);
            assert_eq!(fill.last_px.as_f64(), 3.5);
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_weighted_avg_back_calculation() {
        let mut tracker = FillTracker::new();
        let instrument_id = InstrumentId::from("1.234567-999003-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::from("GBP");
        let ts = UnixNanos::default();

        // First fill: 10 @ avp=2.0
        let uo1 = make_test_uo(
            "999003",
            Decimal::new(30, 0),
            Some(Decimal::new(10, 0)),
            Some(Decimal::new(20, 1)),
        );
        let fill1 = tracker
            .maybe_fill_report(&uo1, uo1.s, instrument_id, account_id, currency, ts, ts)
            .expect("first fill");
        assert_eq!(fill1.last_px.as_f64(), 2.0);
        assert_eq!(fill1.last_qty.as_f64(), 10.0);

        // Second fill: sm=20, avp=2.5
        // Back-calc: (2.5*20 - 2.0*10) / 10 = (50-20)/10 = 3.0
        let uo2 = make_test_uo(
            "999003",
            Decimal::new(30, 0),
            Some(Decimal::new(20, 0)),
            Some(Decimal::new(25, 1)),
        );
        let fill2 = tracker
            .maybe_fill_report(&uo2, uo2.s, instrument_id, account_id, currency, ts, ts)
            .expect("second fill");
        assert_eq!(fill2.last_qty.as_f64(), 10.0);
        assert_eq!(fill2.last_px.as_f64(), 3.0);
    }

    #[rstest]
    fn test_fill_tracker_negative_fill_price_falls_back_to_avp() {
        let mut tracker = FillTracker::new();
        let instrument_id = InstrumentId::from("1.234567-999004-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::from("GBP");
        let ts = UnixNanos::default();

        // First fill: 10 @ avp=5.0
        let uo1 = make_test_uo(
            "999004",
            Decimal::new(20, 0),
            Some(Decimal::new(10, 0)),
            Some(Decimal::new(50, 1)),
        );
        tracker
            .maybe_fill_report(&uo1, uo1.s, instrument_id, account_id, currency, ts, ts)
            .expect("first fill");

        // Second fill: sm=15, avp=1.0
        // Back-calc: (1.0*15 - 5.0*10) / 5 = (15-50)/5 = -7.0
        // Negative price should fall back to avp=1.0
        let uo2 = make_test_uo(
            "999004",
            Decimal::new(20, 0),
            Some(Decimal::new(15, 0)),
            Some(Decimal::new(10, 1)),
        );
        let fill2 = tracker
            .maybe_fill_report(&uo2, uo2.s, instrument_id, account_id, currency, ts, ts)
            .expect("second fill should use avp fallback");
        assert_eq!(fill2.last_qty.as_f64(), 5.0);
        assert_eq!(fill2.last_px.as_f64(), 1.0);
    }

    #[rstest]
    fn test_fill_tracker_prune_clears_state() {
        let mut tracker = FillTracker::new();
        let instrument_id = InstrumentId::from("1.234567-999005-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let currency = Currency::from("GBP");
        let ts = UnixNanos::default();

        // Fill order fully
        let uo = make_test_uo(
            "999005",
            Decimal::new(10, 0),
            Some(Decimal::new(10, 0)),
            Some(Decimal::new(25, 1)),
        );
        let fill1 =
            tracker.maybe_fill_report(&uo, uo.s, instrument_id, account_id, currency, ts, ts);
        assert!(fill1.is_some());

        // Same data again - deduplicated
        let fill2 =
            tracker.maybe_fill_report(&uo, uo.s, instrument_id, account_id, currency, ts, ts);
        assert!(fill2.is_none(), "should be deduplicated");

        // Prune the bet
        tracker.prune("999005");

        // After prune, same data can produce a fill again (simulates re-processing)
        let fill3 =
            tracker.maybe_fill_report(&uo, uo.s, instrument_id, account_id, currency, ts, ts);
        assert!(fill3.is_some(), "after prune, should produce fill again");
    }

    #[rstest]
    fn test_fill_tracker_sm_none_returns_none() {
        let mut tracker = FillTracker::new();

        // sm=None (no matched quantity field at all)
        let uo = make_test_uo("999006", Decimal::new(10, 0), None, None);

        let instrument_id = InstrumentId::from("1.234567-999006-0.0.BETFAIR");
        let result = tracker.maybe_fill_report(
            &uo,
            uo.s,
            instrument_id,
            AccountId::from("BETFAIR-001"),
            Currency::from("GBP"),
            UnixNanos::default(),
            UnixNanos::default(),
        );
        assert!(result.is_none(), "None sm should not produce a fill");
    }

    #[rstest]
    fn test_parse_order_status_report_missing_persistence_type_for_market_on_close() {
        let uo = UnmatchedOrder {
            s: Decimal::ZERO,
            pt: None,
            ot: StreamingOrderType::MarketOnClose,
            sr: Some(Decimal::new(10, 0)),
            ..make_test_uo("999007", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let report = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.quantity, Quantity::from("10.00"));
        assert_eq!(report.time_in_force, TimeInForce::AtTheClose);
    }

    #[rstest]
    fn test_parse_order_status_report_missing_persistence_type_for_limit_on_close() {
        let uo = UnmatchedOrder {
            s: Decimal::ZERO,
            pt: None,
            ot: StreamingOrderType::LimitOnClose,
            sr: Some(Decimal::new(10, 0)),
            ..make_test_uo("999013", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let report = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.quantity, Quantity::from("10.00"));
        assert_eq!(report.time_in_force, TimeInForce::AtTheClose);
    }

    #[rstest]
    fn test_parse_order_status_report_market_on_close_uses_bsp_liability() {
        let uo = UnmatchedOrder {
            s: Decimal::ZERO,
            bsp: Some(Decimal::new(20, 1)),
            pt: None,
            ot: StreamingOrderType::MarketOnClose,
            ..make_test_uo("999010", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let report = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.quantity, Quantity::from("2.00"));
        assert_eq!(report.time_in_force, TimeInForce::AtTheClose);
    }

    #[rstest]
    fn test_parse_order_status_report_fails_for_non_positive_quantity() {
        let uo = UnmatchedOrder {
            s: Decimal::ZERO,
            sm: Some(Decimal::ZERO),
            sr: Some(Decimal::ZERO),
            sc: Some(Decimal::ZERO),
            sl: Some(Decimal::ZERO),
            sv: Some(Decimal::ZERO),
            ..make_test_uo("999014", Decimal::ZERO, Some(Decimal::ZERO), None)
        };

        let result = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        );

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("failed to resolve positive quantity for stream order update 999014")
        );
    }

    #[rstest]
    fn test_parse_order_status_report_includes_lapse_reason() {
        let uo = UnmatchedOrder {
            status: StreamingOrderStatus::ExecutionComplete,
            sl: Some(Decimal::ONE),
            lsrc: Some(crate::common::enums::LapseStatusReasonCode::SpInPlay),
            ..make_test_uo("999012", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let report = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert_eq!(report.order_status, OrderStatus::Canceled);
        assert_eq!(report.cancel_reason.as_deref(), Some("SP_IN_PLAY"));
    }

    #[rstest]
    fn test_parse_order_status_report_blank_rfo_normalizes_to_none() {
        // Some venues send rfo:"" instead of omitting it. ClientOrderId rejects
        // empty strings, so the parser must treat blank refs as missing.
        let uo = UnmatchedOrder {
            rfo: Some(String::new()),
            ..make_test_uo("999013", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let report = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
        .unwrap();

        assert!(report.client_order_id.is_none());
    }

    #[rstest]
    fn test_fill_tracker_blank_rfo_normalizes_to_none() {
        let mut tracker = FillTracker::new();
        let uo = UnmatchedOrder {
            rfo: Some(String::new()),
            sm: Some(Decimal::new(5, 0)),
            avp: Some(Decimal::new(25, 1)),
            ..make_test_uo("999015", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let fill = tracker
            .maybe_fill_report(
                &uo,
                uo.s,
                InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
                AccountId::from("BETFAIR-001"),
                Currency::from("GBP"),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .expect("blank rfo should still produce a fill");

        assert!(fill.client_order_id.is_none());
    }

    #[rstest]
    fn test_fill_tracker_first_seen_void_records_without_correction() {
        let mut tracker = FillTracker::new();
        let uo = UnmatchedOrder {
            status: StreamingOrderStatus::ExecutionComplete,
            sm: Some(Decimal::new(50, 0)),
            sv: Some(Decimal::new(50, 0)),
            ..make_test_uo("999016", Decimal::new(50, 0), None, None)
        };

        let first = tracker.maybe_fill_voids(&uo);
        let duplicate = tracker.maybe_fill_voids(&uo);

        assert!(first.is_empty());
        assert!(duplicate.is_empty());
        assert!(!tracker.has_unseen_fill_void(&uo));
    }

    #[rstest]
    fn test_fill_tracker_emits_only_increased_cumulative_void() {
        let mut tracker = FillTracker::new();
        let instrument_id = InstrumentId::from("1.234567-123456-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let mut uo = make_test_uo(
            "999017",
            Decimal::new(60, 0),
            Some(Decimal::new(60, 0)),
            Some(Decimal::new(25, 1)),
        );
        tracker
            .maybe_fill_report(
                &uo,
                uo.s,
                instrument_id,
                account_id,
                Currency::GBP(),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .expect("matched size should establish a fill lot");

        uo.sv = Some(Decimal::new(40, 0));
        let first = tracker.maybe_fill_voids(&uo);
        let duplicate = tracker.maybe_fill_voids(&uo);
        uo.sv = Some(Decimal::new(50, 0));
        let unseen_before_update = tracker.has_unseen_fill_void(&uo);
        let increased = tracker.maybe_fill_voids(&uo);

        assert_eq!(first.len(), 1);
        assert_eq!(first[0].voided_qty, Quantity::from("40.00"));
        assert!(duplicate.is_empty());
        assert!(unseen_before_update);
        assert_eq!(increased.len(), 1);
        assert_eq!(increased[0].trade_id, first[0].trade_id);
        assert_eq!(increased[0].voided_qty, Quantity::from("50.00"));
        assert!(!tracker.has_unseen_fill_void(&uo));
    }

    #[rstest]
    fn test_fill_tracker_allocates_cumulative_void_to_newest_fill_lot_first() {
        let mut tracker = FillTracker::new();
        let bet_id = "999019";
        let older_trade_id = TradeId::from("999019-30.00");
        let newer_trade_id = TradeId::from("999019-100.00");
        tracker.sync_fill_lot(
            bet_id,
            older_trade_id,
            Decimal::new(30, 0),
            Price::from("2.00"),
            Decimal::ZERO,
        );
        tracker.sync_fill_lot(
            bet_id,
            newer_trade_id,
            Decimal::new(70, 0),
            Price::from("3.00"),
            Decimal::ZERO,
        );
        let uo = UnmatchedOrder {
            sv: Some(Decimal::new(50, 0)),
            ..make_test_uo(bet_id, Decimal::new(100, 0), None, None)
        };

        let corrections = tracker.maybe_fill_voids(&uo);

        assert_eq!(corrections.len(), 1);
        assert_eq!(corrections[0].trade_id, newer_trade_id);
        assert_eq!(corrections[0].voided_qty, Quantity::from("50.00"));
        assert_eq!(corrections[0].last_px, Price::from("3.00"));
    }

    #[rstest]
    fn test_fill_tracker_prices_new_fill_from_post_void_average() {
        let mut tracker = FillTracker::new();
        let bet_id = "999021";
        let older_trade_id = TradeId::from("999021-50.00");
        tracker.sync_fill_lot(
            bet_id,
            older_trade_id,
            Decimal::new(50, 0),
            Price::from("2.00"),
            Decimal::ZERO,
        );
        let (_, last_qty, last_px) = tracker
            .advance_cumulative_fill_with_voids(
                bet_id,
                Decimal::new(100, 0),
                Decimal::new(40, 0),
                Some(Decimal::new(25, 1)),
                Decimal::new(20, 1),
            )
            .expect("gross lifecycle increase should produce a fill");

        let uo = UnmatchedOrder {
            sm: Some(Decimal::new(60, 0)),
            sv: Some(Decimal::new(40, 0)),
            ..make_test_uo(bet_id, Decimal::new(100, 0), None, None)
        };
        let corrections = tracker.maybe_fill_voids(&uo);

        assert_eq!(last_qty, Quantity::from("50.00"));
        assert_eq!(last_px, Price::from("5.00"));
        assert_eq!(corrections.len(), 1);
        assert_eq!(corrections[0].voided_qty, Quantity::from("40.00"));
        assert_eq!(corrections[0].last_px, Price::from("5.00"));
    }

    #[rstest]
    fn test_fill_tracker_preserves_prior_void_allocation_when_new_fill_arrives() {
        let mut tracker = FillTracker::new();
        let bet_id = "999020";
        let older_trade_id = TradeId::from("999020-60.00");
        let newer_trade_id = TradeId::from("999020-80.00");
        tracker.sync_fill_lot(
            bet_id,
            older_trade_id,
            Decimal::new(60, 0),
            Price::from("2.00"),
            Decimal::ZERO,
        );
        let mut uo = UnmatchedOrder {
            sv: Some(Decimal::new(40, 0)),
            ..make_test_uo(bet_id, Decimal::new(80, 0), None, None)
        };

        let first = tracker.maybe_fill_voids(&uo);
        tracker.sync_fill_lot(
            bet_id,
            newer_trade_id,
            Decimal::new(20, 0),
            Price::from("3.00"),
            Decimal::ZERO,
        );
        uo.sv = Some(Decimal::new(50, 0));
        let increased = tracker.maybe_fill_voids(&uo);

        assert_eq!(first.len(), 1);
        assert_eq!(first[0].trade_id, older_trade_id);
        assert_eq!(first[0].voided_qty, Quantity::from("40.00"));
        assert_eq!(increased.len(), 1);
        assert_eq!(increased[0].trade_id, newer_trade_id);
        assert_eq!(increased[0].voided_qty, Quantity::from("10.00"));
    }

    #[rstest]
    fn test_fill_tracker_reconnect_increased_void_does_not_replay_gross_fill() {
        let mut tracker = FillTracker::new();
        let instrument_id = InstrumentId::from("1.234567-123456-0.0.BETFAIR");
        let account_id = AccountId::from("BETFAIR-001");
        let mut uo = make_test_uo(
            "999018",
            Decimal::new(60, 0),
            Some(Decimal::new(60, 0)),
            Some(Decimal::new(25, 1)),
        );
        let trade_id = make_trade_id(&uo);
        tracker.sync_fill_lot(
            &uo.id,
            trade_id,
            Decimal::new(60, 0),
            Price::from("2.50"),
            Decimal::new(40, 0),
        );
        tracker.sync_voided_qty(&uo.id, Decimal::new(40, 0));
        uo.sv = Some(Decimal::new(50, 0));

        let fill = tracker.maybe_fill_report(
            &uo,
            uo.s,
            instrument_id,
            account_id,
            Currency::GBP(),
            UnixNanos::default(),
            UnixNanos::default(),
        );
        let corrections = tracker.maybe_fill_voids(&uo);

        assert!(fill.is_none());
        assert_eq!(corrections.len(), 1);
        assert_eq!(corrections[0].trade_id, trade_id);
        assert_eq!(corrections[0].voided_qty, Quantity::from("50.00"));
    }

    #[rstest]
    fn test_parse_order_status_report_missing_persistence_type_fails_for_limit_order() {
        let uo = UnmatchedOrder {
            pt: None,
            ..make_test_uo("999008", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let result = parse_order_status_report(
            &uo,
            InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
            AccountId::from("BETFAIR-001"),
            UnixNanos::default(),
            UnixNanos::default(),
        );

        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            "missing persistence type for order update 999008"
        );
    }

    #[rstest]
    fn test_fill_tracker_uses_lifecycle_quantity_when_stream_size_is_zero() {
        let mut tracker = FillTracker::new();
        let uo = UnmatchedOrder {
            s: Decimal::ZERO,
            sr: Some(Decimal::new(10, 0)),
            sm: Some(Decimal::new(5, 0)),
            avp: Some(Decimal::new(20, 1)),
            ..make_test_uo("999009", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let fill = tracker
            .maybe_fill_report(
                &uo,
                uo.s,
                InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
                AccountId::from("BETFAIR-001"),
                Currency::from("GBP"),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .expect("zero stream size should fall back to lifecycle quantities");

        assert_eq!(fill.last_qty, Quantity::from("5.00"));
    }

    #[rstest]
    fn test_fill_tracker_uses_bsp_liability_when_stream_size_is_zero() {
        let mut tracker = FillTracker::new();
        let uo = UnmatchedOrder {
            s: Decimal::ZERO,
            bsp: Some(Decimal::new(20, 1)),
            pt: None,
            ot: StreamingOrderType::MarketOnClose,
            sm: Some(Decimal::new(10, 1)),
            avp: Some(Decimal::new(20, 1)),
            ..make_test_uo("999011", Decimal::new(10, 0), Some(Decimal::ZERO), None)
        };

        let fill = tracker
            .maybe_fill_report(
                &uo,
                uo.s,
                InstrumentId::from("1.234567-123456-0.0.BETFAIR"),
                AccountId::from("BETFAIR-001"),
                Currency::from("GBP"),
                UnixNanos::default(),
                UnixNanos::default(),
            )
            .expect("zero stream size should fall back to bsp liability");

        assert_eq!(fill.last_qty, Quantity::from("1.00"));
    }

    #[rstest]
    fn test_fill_tracker_partial_void_still_emits_fill() {
        let data = load_test_json("stream/ocm_VOIDED_partial.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let fill = tracker
                .maybe_fill_report(
                    uo,
                    uo.s,
                    instrument_id,
                    AccountId::from("BETFAIR-001"),
                    Currency::GBP(),
                    ts,
                    ts,
                )
                .expect("should produce fill for matched portion");

            // Order: s=100, sm=60, sv=40 -> fill qty should be 60
            assert_eq!(fill.last_qty.as_f64(), 60.0);
            assert_eq!(fill.last_px.as_f64(), 1.5);
            assert_eq!(fill.order_side, OrderSide::Sell);

            let report = parse_order_status_report(
                uo,
                instrument_id,
                AccountId::from("BETFAIR-001"),
                ts,
                ts,
            )
            .unwrap();
            assert_eq!(report.filled_qty, Quantity::from("60.00"));
        } else {
            panic!("expected OrderChange");
        }
    }

    #[rstest]
    fn test_fill_tracker_no_fill_when_sv_zero_and_fully_filled() {
        let data = load_test_json("stream/ocm_FILLED_sv_zero.json");
        let msg: StreamMessage = serde_json::from_str(&data).unwrap();

        if let StreamMessage::OrderChange(ocm) = msg {
            let oc = ocm.oc.as_ref().unwrap();
            let omc = &oc[0];
            let orc = &omc.orc.as_ref().unwrap()[0];
            let uo = &orc.uo.as_ref().unwrap()[0];
            let instrument_id = make_instrument_id(&omc.id, orc.id, Decimal::ZERO);
            let ts = parse_millis_timestamp(ocm.pt);

            let mut tracker = FillTracker::new();
            let fill = tracker
                .maybe_fill_report(
                    uo,
                    uo.s,
                    instrument_id,
                    AccountId::from("BETFAIR-001"),
                    Currency::GBP(),
                    ts,
                    ts,
                )
                .expect("fully filled order should produce fill");

            assert_eq!(fill.last_qty.as_f64(), 50.0);
            assert_eq!(fill.last_px.as_f64(), 2.0);

            // sv=0, so no void event should be generated (tested separately)
            assert_eq!(uo.sv, Some(Decimal::ZERO));
        } else {
            panic!("expected OrderChange");
        }
    }
}