Optimize BacktestEngine data loading with deferred sorting

- Add `_sorted` flag and `sort=False` parameter to defer sorting
- Add presorted parameter to data iterator to skip redundant sorts
- Optimize `BacktestNode` to sort once after loading all data configs
- Add test coverage and documentation for deferred sorting

Author: cjdsellers

docs/concepts/backtesting.md

@@ -39,6 +39,109 @@ An instantiated `BacktestEngine` can accept the following:
 
 This approach offers detailed control over the backtesting process, allowing you to manually configure each component.
 
+### Loading large datasets efficiently
+
+When working with large amounts of data across multiple instruments, the way you load data
+can significantly impact performance.
+
+#### The performance consideration
+
+By default, `BacktestEngine.add_data()` sorts the entire data stream (existing data + newly
+added data) on each call when `sort=True` (the default). This means:
+
+- First call with 1M bars: sorts 1M bars.
+- Second call with 1M bars: sorts 2M bars.
+- Third call with 1M bars: sorts 3M bars.
+- And so on...
+
+This repeated sorting of increasingly large datasets can become a bottleneck when loading
+data for multiple instruments.
+
+#### Optimization strategies
+
+**Strategy 1: Defer sorting until the end (recommended for multiple instruments)**
+
+```python
+from nautilus_trader.backtest.engine import BacktestEngine
+
+engine = BacktestEngine()
+
+# Setup venue and instruments
+engine.add_venue(...)
+engine.add_instrument(instrument1)
+engine.add_instrument(instrument2)
+engine.add_instrument(instrument3)
+
+# Load all data WITHOUT sorting on each call
+engine.add_data(instrument1_bars, sort=False)
+engine.add_data(instrument2_bars, sort=False)
+engine.add_data(instrument3_bars, sort=False)
+
+# Sort once at the end - much more efficient!
+engine.sort_data()
+
+# Now run your backtest
+engine.add_strategy(strategy)
+engine.run()
+```
+
+**Strategy 2: Collect and add in a single batch**
+
+```python
+# Collect all data first
+all_bars = []
+all_bars.extend(instrument1_bars)
+all_bars.extend(instrument2_bars)
+all_bars.extend(instrument3_bars)
+
+# Add once with sorting
+engine.add_data(all_bars, sort=True)
+```
+
+**Strategy 3: Use streaming API for very large datasets**
+
+For datasets that don't fit in memory, use the streaming API:
+
+```python
+def data_generator():
+    # Yield chunks of pre-sorted data
+    yield load_chunk_1()
+    yield load_chunk_2()
+    yield load_chunk_3()
+
+engine.add_data_iterator(
+    data_name="my_data_stream",
+    generator=data_generator(),
+)
+```
+
+:::tip Performance impact
+For a backtest with 10 instruments, each with 1M bars:
+
+- Sorting on each call: ~10 sorts of increasing size (1M, 2M, 3M, ... 10M bars).
+- Sorting once at the end: 1 sort of 10M bars.
+
+The deferred sorting approach can be **orders of magnitude faster** for large datasets.
+:::
+
+### Data loading contract
+
+The `BacktestEngine` enforces important invariants to ensure data integrity:
+
+**Requirements:**
+
+- All data must be sorted and synced to the internal iterator before calling `run()`.
+- When using `sort=False`, you **must** call `sort_data()` or add more data with `sort=True` before running.
+- The engine validates this requirement and raises `RuntimeError` if violated.
+
+**Safety guarantees:**
+
+- Data lists are always copied internally to prevent external mutations from affecting engine state.
+- You can safely clear or modify data lists after passing them to `add_data()`.
+- Adding data with `sort=True` makes it immediately available for backtesting.
+
+This design ensures data integrity while enabling performance optimizations for large datasets.
+
 ## High-level API
 
 The high-level API centers around a `BacktestNode`, which orchestrates the management of multiple `BacktestEngine` instances,

nautilus_trader/backtest/engine.pxd

@@ -114,6 +114,7 @@ cdef class BacktestEngine:
     cdef object _data_iterator
     cdef uint64_t _last_ns
     cdef uint64_t _end_ns
+    cdef bint _sorted
     cdef dict[str, RequestData] _data_requests
     cdef set[str] _backtest_subscription_names
     cdef dict[str, uint64_t] _last_subscription_ts
@@ -176,7 +177,7 @@ cdef class BacktestDataIterator:
     cdef dict[str, uint64_t] _stream_chunk_duration_ns
 
     cpdef void _reset_single_data(self)
-    cdef void _add_data(self, str data_name, list data_list, bint append_data=*)
+    cdef void _add_data(self, str data_name, list data_list, bint append_data=*, bint presorted=*)
     cpdef void remove_data(self, str data_name, bint complete_remove=*)
     cpdef void _activate_single_data(self)
     cpdef void _deactivate_single_data(self)

nautilus_trader/backtest/engine.pyx

@@ -230,6 +230,7 @@ cdef class BacktestEngine:
         self._iteration: uint64_t = 0
         self._last_ns : uint64_t = 0
         self._end_ns : uint64_t = 0
+        self._sorted: bint = True
 
         # Timing
         self._run_started: pd.Timestamp | None = None
@@ -767,6 +768,30 @@ cdef class BacktestEngine:
         Caution if adding data without `sort` being True, as this could lead to running backtests
         on a stream which does not have monotonically increasing timestamps.
 
+        Notes
+        -----
+        For optimal performance when loading large datasets, consider using `sort=False` for all
+        calls to `add_data()`, then calling `sort_data()` once after all data has been added:
+
+        .. code-block:: python
+
+            # Add multiple data streams without sorting
+            engine.add_data(instrument1_bars, sort=False)
+            engine.add_data(instrument2_bars, sort=False)
+            engine.add_data(instrument3_bars, sort=False)
+
+            # Sort once at the end
+            engine.sort_data()
+
+        This approach avoids repeatedly sorting the entire data stream on each call,
+        significantly reducing load time for large datasets.
+
+        **Contract invariants:**
+
+        - When `sort=True`: Data is immediately available for backtesting via `run()`.
+        - When `sort=False`: You **must** call `sort_data()` or add data with `sort=True` before `run()`.
+        - The provided `data` list is always copied internally to prevent external mutations from affecting the engine state.
+
         """
         Condition.not_empty(data, "data")
         Condition.list_type(data, Data, "data")
@@ -822,8 +847,10 @@ cdef class BacktestEngine:
 
         if sort:
             self._data = sorted(self._data, key=lambda x: x.ts_init)
-
-        self._data_iterator.add_data("backtest_data", self._data)
+            self._data_iterator.add_data("backtest_data", self._data, append_data=True, presorted=True)
+            self._sorted = True
+        else:
+            self._sorted = False
 
         for data_point in data:
             data_type = type(data_point)
@@ -1049,6 +1076,8 @@ cdef class BacktestEngine:
         """
         Condition.not_none(data, "data")
         self._data = pickle.loads(data)
+        self._data_iterator.add_data("backtest_data", self._data, append_data=True, presorted=True)
+        self._sorted = True
 
         self._log.info(
             f"Loaded {len(self._data):_} data "
@@ -1191,7 +1220,10 @@ cdef class BacktestEngine:
         # Reset timing
         self._iteration = 0
         self._data_iterator = BacktestDataIterator()
-        self._data_iterator.add_data("backtest_data", self._data)
+
+        if self._sorted:
+            self._data_iterator.add_data("backtest_data", self._data, append_data=True, presorted=True)
+
         self._run_started = None
         self._run_finished = None
         self._backtest_start = None
@@ -1204,7 +1236,9 @@ cdef class BacktestEngine:
         Sort the engines internal data stream.
 
         """
-        self._data.sort()
+        self._data = sorted(self._data, key=lambda x: x.ts_init)
+        self._data_iterator.add_data("backtest_data", self._data, append_data=True, presorted=True)
+        self._sorted = True
 
     def clear_data(self) -> None:
         """
@@ -1218,6 +1252,7 @@ cdef class BacktestEngine:
         self._data.clear()
         self._data_len = 0
         self._data_iterator = BacktestDataIterator()
+        self._sorted = True
 
     def clear_actors(self) -> None:
         """
@@ -1294,6 +1329,16 @@ cdef class BacktestEngine:
             If no data has been added to the engine.
         ValueError
             If the `start` is >= the `end` datetime.
+        RuntimeError
+            If data has been added with `sort=False` but `sort_data()` has not been called.
+
+        Notes
+        -----
+        **Contract invariants:**
+
+        - All data added via `add_data()` must be sorted and synced to the internal iterator before calling `run()`.
+        - If any data was added with `sort=False`, you must call `sort_data()` or add data with `sort=True` before this method.
+        - The engine validates this requirement and will raise `RuntimeError` if unsorted data is detected.
 
         """
         self._run(start, end, run_config_id, streaming)
@@ -1393,6 +1438,13 @@ cdef class BacktestEngine:
         run_config_id: str | None = None,
         bint streaming = False,
     ):
+        # Validate data has been sorted and synced to iterator
+        if self._data and not self._sorted:
+            raise RuntimeError(
+                "Data has been added but not sorted, "
+                "call `engine.sort_data()` or use `engine.add_data(..., sort=True)` before running"
+            )
+
         # Validate data
         cdef:
             SimulatedExchange exchange
@@ -1962,7 +2014,13 @@ cdef class BacktestDataIterator:
         self._single_data_index = 0
         self._is_single_data = False
 
-    def add_data(self, data_name, list data, bint append_data=True):
+    def add_data(
+        self,
+        str data_name,
+        list data,
+        bint append_data = True,
+        bint presorted = False,
+    ) -> None:
         """
         Add (or replace) a named, pre-sorted data list for static data loading.
 
@@ -1979,6 +2037,9 @@ cdef class BacktestDataIterator:
             Controls stream priority for timestamp ties:
             ``True`` – lower priority (appended).
             ``False`` – higher priority (prepended).
+        presorted : bool, default ``False``
+            If the data is guaranteed to be pre-sorted by `ts_init`.
+            When ``True``, skips internal sorting for better performance.
 
         Raises
         ------
@@ -1991,13 +2052,14 @@ cdef class BacktestDataIterator:
         if not data:
             return
 
-        def data_generator():
-            yield data
-            # Generator ends after yielding once
-
-        self.init_data(data_name, data_generator(), append_data)
+        self._add_data(data_name, data, append_data, presorted)
 
-    def init_data(self, str data_name, data_generator, bint append_data=True):
+    def init_data(
+        self,
+        str data_name,
+        data_generator,
+        bint append_data = True,
+    ) -> None:
         """
         Add (or replace) a named data generator for streaming large datasets.
 
@@ -2042,7 +2104,13 @@ cdef class BacktestDataIterator:
             # Generator is already exhausted, nothing to add
             pass
 
-    cdef void _add_data(self, str data_name, list data_list, bint append_data=True):
+    cdef void _add_data(
+        self,
+        str data_name,
+        list data_list,
+        bint append_data = True,
+        bint presorted = False,
+    ):
         if len(data_list) == 0:
             return
 
@@ -2062,7 +2130,12 @@ cdef class BacktestDataIterator:
         if self._is_single_data:
             self._deactivate_single_data()
 
-        self._data[data_priority] = sorted(data_list, key=lambda data: data.ts_init)
+        # Copy and optionally sort to avoid aliasing caller's list
+        if presorted:
+            self._data[data_priority] = list(data_list)
+        else:
+            self._data[data_priority] = sorted(data_list, key=lambda data: data.ts_init)
+
         self._data_name[data_priority] = data_name
         self._data_priority[data_name] = data_priority
         self._data_len[data_priority] = len(data_list)

nautilus_trader/backtest/node.py

@@ -592,7 +592,7 @@ def _run_oneshot(
         start: str | int | None = None,
         end: str | int | None = None,
     ) -> None:
-        # Load data
+        # Load data - defer sorting until all data is loaded for better performance
         for config in data_configs:
             t0 = pd.Timestamp.now()
             used_instrument_ids = get_instrument_ids(config)
@@ -616,11 +616,12 @@ def _run_oneshot(
                 f"Read {len(result.data):,} events from parquet in {pd.Timedelta(t1 - t0)}s",
             )
 
-            self._load_engine_data(engine=engine, result=result)
+            self._load_engine_data(engine=engine, result=result, sort=False)  # sort before run
 
             t2 = pd.Timestamp.now()
             engine.logger.info(f"Engine load took {pd.Timedelta(t2 - t1)}s")
 
+        engine.sort_data()
         engine.run(start=start, end=end, run_config_id=run_config_id)
 
     @classmethod
@@ -669,11 +670,16 @@ def load_catalog(cls, config: BacktestDataConfig) -> ParquetDataCatalog:
             fs_rust_storage_options=config.catalog_fs_rust_storage_options,
         )
 
-    def _load_engine_data(self, engine: BacktestEngine, result: CatalogDataResult) -> None:
+    def _load_engine_data(
+        self,
+        engine: BacktestEngine,
+        result: CatalogDataResult,
+        sort: bool = True,
+    ) -> None:
         if is_nautilus_class(result.data_cls):
             engine.add_data(
                 data=result.data,
-                sort=True,  # Already sorted from backend
+                sort=sort,
             )
         else:
             if not result.client_id:
@@ -684,7 +690,7 @@ def _load_engine_data(self, engine: BacktestEngine, result: CatalogDataResult) -
             engine.add_data(
                 data=result.data,
                 client_id=result.client_id,
-                sort=True,  # Already sorted from backend
+                sort=sort,
             )
 
     def log_backtest_exception(self, e: Exception, config: BacktestRunConfig) -> None: