nautilus_backtest/

node.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.
// -------------------------------------------------------------------------------------------------

//! Provides a [`BacktestNode`] that orchestrates catalog-driven backtests.

use std::iter::Peekable;

use ahash::{AHashMap, AHashSet};
use nautilus_core::UnixNanos;
use nautilus_model::{
    data::{
        Bar, Data, FundingRateUpdate, HasTsInit, IndexPriceUpdate, InstrumentClose,
        InstrumentStatus, MarkPriceUpdate, OptionGreeks, OrderBookDelta, OrderBookDepth10,
        QuoteTick, TradeTick,
    },
    enums::{BookType, OtoTriggerMode},
    identifiers::{InstrumentId, Venue},
    types::Money,
};
use nautilus_persistence::backend::{catalog::ParquetDataCatalog, session::QueryResult};

use crate::{
    config::{BacktestDataConfig, BacktestRunConfig, NautilusDataType, SimulatedVenueConfig},
    engine::BacktestEngine,
    result::BacktestResult,
};

/// Orchestrates catalog-driven backtests from run configurations.
///
/// `BacktestNode` connects the [`ParquetDataCatalog`] with [`BacktestEngine`] to load
/// historical data and run backtests. Supports both oneshot and streaming modes.
#[derive(Debug)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.backtest", unsendable)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.backtest")
)]
pub struct BacktestNode {
    configs: Vec<BacktestRunConfig>,
    engines: AHashMap<String, BacktestEngine>,
}

impl BacktestNode {
    /// Creates a new [`BacktestNode`] instance.
    ///
    /// Validates that configs are non-empty and internally consistent:
    /// - All data config instrument venues must have a matching venue config.
    /// - L2/L3 book types require order book data in the data configs.
    /// - Data config time ranges must be valid (start <= end).
    ///
    /// # Errors
    ///
    /// Returns an error if `configs` is empty or validation fails.
    pub fn new(configs: Vec<BacktestRunConfig>) -> anyhow::Result<Self> {
        anyhow::ensure!(!configs.is_empty(), "At least one run config is required");
        validate_configs(&configs)?;
        Ok(Self {
            configs,
            engines: AHashMap::new(),
        })
    }

    /// Returns the run configurations.
    #[must_use]
    pub fn configs(&self) -> &[BacktestRunConfig] {
        &self.configs
    }

    /// Builds backtest engines from the run configurations.
    ///
    /// For each config, creates a [`BacktestEngine`], adds venues, and loads
    /// instruments from the catalog. If building a config fails with
    /// [`BacktestRunConfig::raise_exception`] disabled, logs the error and skips that config;
    /// successful return does not guarantee an engine for every config.
    ///
    /// # Errors
    ///
    /// Returns an error if building an engine from a config fails and
    /// [`BacktestRunConfig::raise_exception`] is enabled for that config.
    pub fn build(&mut self) -> anyhow::Result<()> {
        for config in &self.configs {
            if self.engines.contains_key(config.id()) {
                continue;
            }

            match build_engine(config) {
                Ok(engine) => {
                    self.engines.insert(config.id().to_string(), engine);
                }
                Err(e) if config.raise_exception() => return Err(e),
                Err(e) => {
                    log::error!("Error building backtest '{}': {e:#}", config.id());
                }
            }
        }

        Ok(())
    }

    /// Returns a mutable reference to the engine for the given run config ID.
    #[must_use]
    pub fn get_engine_mut(&mut self, id: &str) -> Option<&mut BacktestEngine> {
        self.engines.get_mut(id)
    }

    /// Returns a reference to the engine for the given run config ID.
    #[must_use]
    pub fn get_engine(&self, id: &str) -> Option<&BacktestEngine> {
        self.engines.get(id)
    }

    /// Returns all created backtest engines.
    #[must_use]
    pub fn get_engines(&self) -> Vec<&BacktestEngine> {
        self.engines.values().collect()
    }

    /// Runs all configured backtests and returns results.
    ///
    /// Automatically calls [`build()`](Self::build) if engines have not been created yet.
    /// For each run config, loads data from the catalog and runs the engine.
    /// Supports both oneshot (`chunk_size = None`) and streaming modes.
    /// Configs without a built engine are skipped. If a run fails with
    /// [`BacktestRunConfig::raise_exception`] disabled, logs the error, clears its loaded data,
    /// leaves the engine undisposed, and omits its result.
    ///
    /// # Errors
    ///
    /// Returns an error if building, data loading, or engine execution fails and
    /// [`BacktestRunConfig::raise_exception`] is enabled for the run config.
    pub fn run(&mut self) -> anyhow::Result<Vec<BacktestResult>> {
        // Auto-build if not already done
        if self.engines.is_empty() {
            self.build()?;
        }

        let mut results = Vec::new();

        for config in &self.configs {
            let Some(engine) = self.engines.get_mut(config.id()) else {
                continue;
            };

            let run_result = match config.chunk_size() {
                None => run_oneshot(engine, config),
                Some(chunk_size) => run_streaming(engine, config, chunk_size),
            };

            if let Err(e) = run_result {
                if config.raise_exception() {
                    return Err(e);
                }

                log::error!("Error running backtest '{}': {e:#}", config.id());
                engine.clear_data();
                continue;
            }

            results.push(engine.get_result());

            if config.dispose_on_completion() {
                engine.dispose();
            } else {
                engine.clear_data();
            }
        }

        Ok(results)
    }

    /// Creates a [`ParquetDataCatalog`] from a data config.
    ///
    /// # Errors
    ///
    /// Returns an error if the catalog cannot be created from the URI.
    pub fn load_catalog(config: &BacktestDataConfig) -> anyhow::Result<ParquetDataCatalog> {
        create_catalog(config)
    }

    /// Loads data from the catalog for a specific data config.
    ///
    /// # Errors
    ///
    /// Returns an error if catalog creation or data querying fails.
    pub fn load_data_config(
        config: &BacktestDataConfig,
        start: Option<UnixNanos>,
        end: Option<UnixNanos>,
    ) -> anyhow::Result<Vec<Data>> {
        load_data(config, start, end)
    }

    /// Disposes all engines and releases resources.
    pub fn dispose(&mut self) {
        for engine in self.engines.values_mut() {
            engine.dispose();
        }
        self.engines.clear();
    }
}

fn build_engine(config: &BacktestRunConfig) -> anyhow::Result<BacktestEngine> {
    let engine_config = config.engine().clone();
    let mut engine = BacktestEngine::new(engine_config)?;

    for venue_config in config.venues() {
        let starting_balances: Vec<Money> = venue_config
            .starting_balances()
            .iter()
            .map(|s| s.parse::<Money>())
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| anyhow::anyhow!("Invalid starting balance: {e}"))?;

        let default_leverage = venue_config.default_leverage();
        let leverages = venue_config.leverages().cloned().unwrap_or_default();
        let margin_model = venue_config.margin_model().cloned().map(Into::into);
        let modules = venue_config
            .modules()
            .iter()
            .cloned()
            .map(Into::into)
            .collect();
        let fill_model = venue_config
            .fill_model()
            .cloned()
            .unwrap_or_default()
            .into();
        let fee_model = venue_config.fee_model().cloned().unwrap_or_default().into();
        let latency_model = venue_config.latency_model().cloned().map(Into::into);
        let sim_config = SimulatedVenueConfig::builder()
            .venue(Venue::from(venue_config.name().as_str()))
            .oms_type(venue_config.oms_type())
            .account_type(venue_config.account_type())
            .book_type(venue_config.book_type())
            .starting_balances(starting_balances)
            .maybe_base_currency(venue_config.base_currency())
            .maybe_default_leverage(default_leverage)
            .leverages(leverages)
            .maybe_margin_model(margin_model)
            .modules(modules)
            .fill_model(fill_model)
            .fee_model(fee_model)
            .maybe_latency_model(latency_model)
            .routing(venue_config.routing())
            .reject_stop_orders(venue_config.reject_stop_orders())
            .support_gtd_orders(venue_config.support_gtd_orders())
            .support_contingent_orders(venue_config.support_contingent_orders())
            .use_position_ids(venue_config.use_position_ids())
            .use_random_ids(venue_config.use_random_ids())
            .use_reduce_only(venue_config.use_reduce_only())
            .use_market_order_acks(venue_config.use_market_order_acks())
            .bar_execution(venue_config.bar_execution())
            .bar_adaptive_high_low_ordering(venue_config.bar_adaptive_high_low_ordering())
            .trade_execution(venue_config.trade_execution())
            .liquidity_consumption(venue_config.liquidity_consumption())
            .allow_cash_borrowing(venue_config.allow_cash_borrowing())
            .frozen_account(venue_config.frozen_account())
            .queue_position(venue_config.queue_position())
            .oto_full_trigger(venue_config.oto_trigger_mode() == OtoTriggerMode::Full)
            .price_protection_points(venue_config.price_protection_points())
            .liquidation_enabled(venue_config.liquidation_enabled())
            .liquidation_trigger_ratio(venue_config.liquidation_trigger_ratio())
            .liquidation_cancel_open_orders(venue_config.liquidation_cancel_open_orders())
            .build()?;
        engine.add_venue(sim_config)?;
    }

    for data_config in config.data() {
        let catalog = create_catalog(data_config)?;
        let instr_ids: Vec<InstrumentId> = data_config.get_instrument_ids()?;
        let filter: Option<Vec<String>> = if instr_ids.is_empty() {
            None
        } else {
            Some(instr_ids.iter().map(ToString::to_string).collect())
        };

        let instruments = catalog.query_instruments(filter.as_deref())?;

        if !instr_ids.is_empty() && instruments.is_empty() {
            let ids: Vec<String> = instr_ids.iter().map(ToString::to_string).collect();
            anyhow::bail!(
                "No instruments found in catalog for requested IDs: [{}]",
                ids.join(", ")
            );
        }

        for instrument in instruments {
            engine.add_instrument(&instrument)?;
        }
    }

    Ok(engine)
}

fn validate_configs(configs: &[BacktestRunConfig]) -> anyhow::Result<()> {
    // Kernel initialization sets a thread-local MessageBus that can only be
    // initialized once per thread, so multiple engines cannot coexist
    anyhow::ensure!(
        configs.len() <= 1,
        "Only one run config per BacktestNode is supported \
         (kernel MessageBus is a thread-local singleton)"
    );

    let mut seen_ids = AHashSet::new();

    for config in configs {
        anyhow::ensure!(
            seen_ids.insert(config.id()),
            "Duplicate run config ID '{}'",
            config.id()
        );

        let venue_names: Vec<String> = config
            .venues()
            .iter()
            .map(|v| v.name().to_string())
            .collect();

        for data_config in config.data() {
            if let (Some(start), Some(end)) = (data_config.start_time(), data_config.end_time()) {
                anyhow::ensure!(
                    start <= end,
                    "Data config start_time ({start}) must be <= end_time ({end})"
                );
            }

            for instrument_id in data_config.get_instrument_ids()? {
                let venue = instrument_id.venue.to_string();
                anyhow::ensure!(
                    venue_names.contains(&venue),
                    "No venue config found for venue '{venue}' (required by instrument {instrument_id})"
                );
            }
        }

        for venue_config in config.venues() {
            let needs_book_data = matches!(
                venue_config.book_type(),
                BookType::L2_MBP | BookType::L3_MBO
            );

            if needs_book_data {
                let venue_name = venue_config.name().to_string();
                let has_book_data = config.data().iter().any(|dc| {
                    let is_book_type = matches!(
                        dc.data_type(),
                        NautilusDataType::OrderBookDelta | NautilusDataType::OrderBookDepth10
                    );

                    if !is_book_type {
                        return false;
                    }

                    // Unfiltered config (no instrument filter) covers all venues
                    let ids = dc.get_instrument_ids().unwrap_or_default();
                    ids.is_empty() || ids.iter().any(|id| id.venue.to_string() == venue_name)
                });
                anyhow::ensure!(
                    has_book_data,
                    "Venue '{venue_name}' has book_type {:?} but no order book data configured",
                    venue_config.book_type()
                );
            }
        }
    }
    Ok(())
}

fn run_oneshot(engine: &mut BacktestEngine, config: &BacktestRunConfig) -> anyhow::Result<()> {
    for data_config in config.data() {
        let data = load_data(data_config, config.start(), config.end())?;
        if data.is_empty() {
            log::warn!("No data found for config: {:?}", data_config.data_type());
            continue;
        }
        engine.add_data(data, data_config.client_id(), false, false)?;
    }

    engine.sort_data();
    engine.run(
        config.start(),
        config.end(),
        Some(config.id().to_string()),
        false,
    )
}

fn run_streaming(
    engine: &mut BacktestEngine,
    config: &BacktestRunConfig,
    chunk_size: usize,
) -> anyhow::Result<()> {
    let data_configs = config.data();

    // Stream directly from the catalog iterators without materializing the full
    // dataset, so memory stays bounded by chunk_size for any number of configs
    let mut catalogs = data_configs
        .iter()
        .map(create_catalog)
        .collect::<anyhow::Result<Vec<_>>>()?;
    let mut streams = Vec::with_capacity(catalogs.len());

    for (catalog, data_config) in catalogs.iter_mut().zip(data_configs) {
        let result = dispatch_query(catalog, data_config, config.start(), config.end())?;
        let mut stream = result
            .map(|item| item.map_err(anyhow::Error::from))
            .peekable();

        match stream.peek() {
            Some(Ok(_)) => streams.push(stream),
            // Surface a failed query in config order, before opening later ones
            Some(Err(_)) => {
                stream.next().transpose()?;
            }
            None => log::warn!("No data found for config: {:?}", data_config.data_type()),
        }
    }

    stream_chunks(
        engine,
        config,
        merge_streams(streams).peekable(),
        chunk_size,
    )
}

// Merges the data streams of every config in ascending `ts_init` order, taking one
// item at a time so the merge holds only a single item per config. Ties keep config
// order, matching the stable sort the eager path applies.
fn merge_streams<I: Iterator<Item = anyhow::Result<Data>>>(
    mut streams: Vec<Peekable<I>>,
) -> impl Iterator<Item = anyhow::Result<Data>> {
    std::iter::from_fn(move || {
        let mut next: Option<(usize, UnixNanos)> = None;

        for (i, stream) in streams.iter_mut().enumerate() {
            match stream.peek() {
                Some(Ok(data)) => {
                    let ts_init = data.ts_init();
                    if next.is_none_or(|(_, ts)| ts_init < ts) {
                        next = Some((i, ts_init));
                    }
                }
                Some(Err(_)) => return stream.next(),
                None => {}
            }
        }

        streams[next?.0].next()
    })
}

// Feeds data from an iterator to the engine in timestamp-aligned chunks.
// Each chunk contains up to `chunk_size` events, extended to include all
// events sharing the boundary timestamp so timers flush correctly.
fn stream_chunks<I: Iterator<Item = anyhow::Result<Data>>>(
    engine: &mut BacktestEngine,
    config: &BacktestRunConfig,
    mut iter: Peekable<I>,
    chunk_size: usize,
) -> anyhow::Result<()> {
    if iter.peek().is_none() {
        return engine.end();
    }

    let mut next_start = config.start();

    loop {
        let chunk = take_aligned_chunk(&mut iter, chunk_size)?;
        if chunk.is_empty() {
            break;
        }

        let is_last = iter.peek().is_none();
        let end = if is_last {
            config.end()
        } else {
            chunk.last().map(HasTsInit::ts_init)
        };

        engine.add_data(chunk, None, false, true)?;
        engine.run(next_start, end, Some(config.id().to_string()), true)?;
        engine.clear_data();

        // A shutdown request during the chunk already triggered end() inside
        // engine.run(); stop loading further chunks so later data is not processed
        if engine.kernel().is_shutdown_requested() {
            return Ok(());
        }

        // Carry forward the end timestamp so the next chunk's run_impl
        // sets clocks contiguously and processes gap timers correctly
        next_start = end;
    }

    engine.end()
}

// Takes up to `chunk_size` items, then extends to include all remaining
// items sharing the boundary timestamp to avoid splitting same-ts events.
fn take_aligned_chunk<I: Iterator<Item = anyhow::Result<Data>>>(
    iter: &mut Peekable<I>,
    chunk_size: usize,
) -> anyhow::Result<Vec<Data>> {
    let mut chunk = Vec::with_capacity(chunk_size);

    for _ in 0..chunk_size {
        match iter.next() {
            Some(item) => chunk.push(item?),
            None => return Ok(chunk),
        }
    }

    if let Some(boundary_ts) = chunk.last().map(HasTsInit::ts_init) {
        // A failing item ends the extension and surfaces on the next chunk
        while let Some(item) = iter.next_if(|item| {
            item.as_ref()
                .is_ok_and(|data| data.ts_init() == boundary_ts)
        }) {
            chunk.push(item?);
        }
    }

    Ok(chunk)
}

fn create_catalog(config: &BacktestDataConfig) -> anyhow::Result<ParquetDataCatalog> {
    let uri = match config.catalog_fs_protocol() {
        Some(protocol) => format!("{protocol}://{}", config.catalog_path()),
        None => config.catalog_path().to_string(),
    };
    let storage_options = config
        .catalog_fs_rust_storage_options()
        .cloned()
        .or_else(|| config.catalog_fs_storage_options().cloned());
    ParquetDataCatalog::from_uri(&uri, storage_options, None, None, None)
}

fn load_data(
    config: &BacktestDataConfig,
    run_start: Option<UnixNanos>,
    run_end: Option<UnixNanos>,
) -> anyhow::Result<Vec<Data>> {
    let mut catalog = create_catalog(config)?;
    let result = dispatch_query(&mut catalog, config, run_start, run_end)?;
    Ok(result.collect::<Result<Vec<_>, _>>()?)
}

fn dispatch_query(
    catalog: &mut ParquetDataCatalog,
    config: &BacktestDataConfig,
    run_start: Option<UnixNanos>,
    run_end: Option<UnixNanos>,
) -> anyhow::Result<QueryResult> {
    catalog.reset_session();

    let identifiers = config.query_identifiers();
    let start = max_opt(config.start_time(), run_start);
    let end = min_opt(config.end_time(), run_end);
    let filter = config.filter_expr();
    let optimize = config.optimize_file_loading();

    #[rustfmt::skip]
    let result = match config.data_type() {
        NautilusDataType::QuoteTick => catalog.query::<QuoteTick>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::TradeTick => catalog.query::<TradeTick>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::Bar => catalog.query::<Bar>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::OrderBookDelta => catalog.query::<OrderBookDelta>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::OrderBookDepth10 => catalog.query::<OrderBookDepth10>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::MarkPriceUpdate => catalog.query::<MarkPriceUpdate>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::IndexPriceUpdate => catalog.query::<IndexPriceUpdate>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::FundingRateUpdate => catalog.query::<FundingRateUpdate>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::InstrumentStatus => catalog.query::<InstrumentStatus>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::OptionGreeks => catalog.query::<OptionGreeks>(identifiers, start, end, filter, None, optimize),
        NautilusDataType::InstrumentClose => catalog.query::<InstrumentClose>(identifiers, start, end, filter, None, optimize),
    };
    result
}

fn max_opt(a: Option<UnixNanos>, b: Option<UnixNanos>) -> Option<UnixNanos> {
    match (a, b) {
        (Some(a), Some(b)) => Some(a.max(b)),
        (Some(a), None) => Some(a),
        (None, Some(b)) => Some(b),
        (None, None) => None,
    }
}

fn min_opt(a: Option<UnixNanos>, b: Option<UnixNanos>) -> Option<UnixNanos> {
    match (a, b) {
        (Some(a), Some(b)) => Some(a.min(b)),
        (Some(a), None) => Some(a),
        (None, Some(b)) => Some(b),
        (None, None) => None,
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "python")]
    use nautilus_model::enums::{AccountType, OmsType};
    use nautilus_model::{
        enums::AggressorSide,
        identifiers::{InstrumentId, TradeId},
        types::{Price, Quantity},
    };
    #[cfg(feature = "python")]
    use pyo3::{ffi::c_str, prelude::*, types::PyDict};
    use rstest::rstest;

    use super::*;
    use crate::config::MAX_BACKTEST_CHUNK_SIZE;
    #[cfg(feature = "python")]
    use crate::{
        config::BacktestVenueConfig,
        modules::SimulationModuleAny,
        python::modules::{PySimulationModule, PythonSimulationModule},
    };

    fn quote(ts_init: u64) -> Data {
        Data::Quote(QuoteTick::new(
            InstrumentId::from("EUR/USD.SIM"),
            Price::from("1.0001"),
            Price::from("1.0002"),
            Quantity::from("100"),
            Quantity::from("100"),
            UnixNanos::from(ts_init),
            UnixNanos::from(ts_init),
        ))
    }

    fn trade(ts_init: u64) -> Data {
        Data::Trade(TradeTick::new(
            InstrumentId::from("EUR/USD.SIM"),
            Price::from("1.0001"),
            Quantity::from("100"),
            AggressorSide::Buy,
            TradeId::from("T-1"),
            UnixNanos::from(ts_init),
            UnixNanos::from(ts_init),
        ))
    }

    fn stream_failure() -> anyhow::Error {
        anyhow::anyhow!("injected stream failure")
    }

    #[rstest]
    fn merge_streams_orders_items_across_streams_by_ts_init() {
        let streams = vec![
            vec![Ok(quote(1)), Ok(quote(3)), Ok(quote(3))]
                .into_iter()
                .peekable(),
            vec![Ok(trade(2)), Ok(trade(3))].into_iter().peekable(),
            vec![].into_iter().peekable(),
        ];

        let merged: Vec<(u64, bool)> = merge_streams(streams)
            .map(|item| item.expect("the merged stream must not fail"))
            .map(|data| (data.ts_init().as_u64(), matches!(data, Data::Trade(_))))
            .collect();

        assert_eq!(
            merged,
            vec![(1, false), (2, true), (3, false), (3, false), (3, true)]
        );
    }

    #[rstest]
    fn merge_streams_leaves_its_streams_undrained() {
        // Unbounded streams, so a merge that materialized its input would never return
        let ok_quote: fn(u64) -> anyhow::Result<Data> = |ts_init| Ok(quote(ts_init));
        let evens = (0u64..).step_by(2).map(ok_quote);
        let odds = (1u64..).step_by(2).map(ok_quote);

        let merged: Vec<u64> = merge_streams(vec![evens.peekable(), odds.peekable()])
            .take(4)
            .map(|item| item.expect("the merged stream must not fail").ts_init())
            .map(|ts_init| ts_init.as_u64())
            .collect();

        assert_eq!(merged, vec![0, 1, 2, 3]);
    }

    #[rstest]
    fn merge_streams_reports_a_stream_failure() {
        let streams = vec![
            vec![Ok(quote(1)), Err(stream_failure())]
                .into_iter()
                .peekable(),
            vec![Ok(quote(2))].into_iter().peekable(),
        ];
        let mut merged = merge_streams(streams);

        let first = merged.next().expect("the first item must be present");
        let second = merged.next().expect("the failure must be yielded");

        assert_eq!(
            first.expect("the first item must not fail").ts_init(),
            UnixNanos::from(1)
        );
        assert_eq!(
            second
                .expect_err("a failed stream must not read as exhaustion")
                .to_string(),
            "injected stream failure"
        );
    }

    #[rstest]
    fn take_aligned_chunk_reports_a_stream_failure() {
        let mut iter = vec![Ok(quote(1)), Err(stream_failure())]
            .into_iter()
            .peekable();

        let chunk = take_aligned_chunk(&mut iter, 4);

        assert_eq!(
            chunk
                .expect_err("a failed stream must not read as a short chunk")
                .to_string(),
            "injected stream failure"
        );
    }

    #[rstest]
    fn take_aligned_chunk_reports_a_failure_found_at_the_boundary() {
        let mut iter = vec![Ok(quote(1)), Err(stream_failure()), Ok(quote(1))]
            .into_iter()
            .peekable();

        let first = take_aligned_chunk(&mut iter, 1).expect("the first chunk must be complete");
        let second = take_aligned_chunk(&mut iter, 1);

        assert_eq!(first.len(), 1);
        assert_eq!(first[0].ts_init(), UnixNanos::from(1));
        assert_eq!(
            second
                .expect_err("the failure must survive the boundary extension")
                .to_string(),
            "injected stream failure"
        );
    }

    #[rstest]
    fn take_aligned_chunk_extends_past_the_boundary_for_equal_timestamps() {
        let mut iter = vec![Ok(quote(1)), Ok(quote(1)), Ok(quote(2))]
            .into_iter()
            .peekable();

        let chunk = take_aligned_chunk(&mut iter, 1).expect("the chunk must be complete");

        assert_eq!(chunk.len(), 2);
        assert_eq!(chunk[0].ts_init(), UnixNanos::from(1));
        assert_eq!(chunk[1].ts_init(), UnixNanos::from(1));
    }

    #[rstest]
    fn take_aligned_chunk_reserves_maximum_supported_capacity() {
        let mut iter = vec![Ok(quote(1))].into_iter().peekable();

        let chunk = take_aligned_chunk(&mut iter, MAX_BACKTEST_CHUNK_SIZE).unwrap();

        assert_eq!(chunk.len(), 1);
        assert!(chunk.capacity() >= MAX_BACKTEST_CHUNK_SIZE);
        assert_eq!(chunk[0].ts_init(), UnixNanos::from(1));
    }

    #[cfg(feature = "python")]
    #[rstest]
    fn build_engine_accepts_python_module_from_node_config() {
        Python::initialize();

        Python::attach(|py| {
            let locals = PyDict::new(py);
            locals
                .set_item("SimulationModule", py.get_type::<PySimulationModule>())
                .unwrap();
            let module = py
                .eval(
                    c_str!(
                        "type('NodeSimulationModule', (SimulationModule,), {\
                            'process': lambda self, ts_now, context: \
                                (setattr(self, 'calls', self.calls + 1), [])[1]\
                        })()"
                    ),
                    None,
                    Some(&locals),
                )
                .unwrap();
            module.setattr("calls", 0).unwrap();

            let venue = BacktestVenueConfig::builder()
                .name("SIM")
                .oms_type(OmsType::Netting)
                .account_type(AccountType::Margin)
                .book_type(BookType::L1_MBP)
                .starting_balances(vec!["1000 USD".to_string()])
                .modules(vec![SimulationModuleAny::Python(
                    PythonSimulationModule::new(module.clone().unbind()),
                )])
                .build()
                .unwrap();
            let config = BacktestRunConfig::builder()
                .venues(vec![venue])
                .data(Vec::new())
                .build()
                .unwrap();
            let mut engine = build_engine(&config).unwrap();

            engine.run(None, None, None, false).unwrap();

            assert_eq!(
                module.getattr("calls").unwrap().extract::<u32>().unwrap(),
                1
            );
        });
    }
}