Add boundary advance helper

docs/benchmarks.md

@@ -93,5 +93,9 @@ The first engine primitive for that design is
 until the first passive fill boundary and returns the event count needed to
 resume replay at the correct row.
 
+The corresponding Python helper is `advance_until_fill_boundary(...)`. It keeps
+the boundary contract testable on the scalar reference engine and the compiled
+engine, which is the step before changing the ordinary replay loop.
+
 CI should keep benchmark code runnable; it should not enforce fixed speed
 thresholds across hardware.

docs/execution-engines.md

@@ -116,6 +116,12 @@ This primitive does not change ordinary audited `Replay(...)` yet. It exists so
 the faster path can be wired in carefully without weakening replay
 inspectability.
 
+`advance_until_fill_boundary(...)` is the Python-side helper that wraps this
+idea. It uses the compiled boundary method when the engine and columns support
+it, and otherwise falls back to scalar `apply_event(...)` calls. That lets tests
+prove the same boundary behavior through the readable Python path and the C++
+path before replay integration depends on it.
+
 Install a source checkout normally to build the extension:
 
 ```bash

src/ordersim/__init__.py

@@ -31,7 +31,14 @@
     default_latency_model_factory,
 )
 from ordersim.recording import RecordingGateway
-from ordersim.replay import CompiledEventColumns, Replay, ReplayGateway, ReplayResult
+from ordersim.replay import (
+    BoundaryAdvance,
+    CompiledEventColumns,
+    Replay,
+    ReplayGateway,
+    ReplayResult,
+    advance_until_fill_boundary,
+)
 from ordersim.sim import (
     CppMatchingEngine,
     ExecutionEngine,
@@ -59,6 +66,7 @@
 
 __all__ = [
     "BookSide",
+    "BoundaryAdvance",
     "CompiledEventColumns",
     "ConstantLatency",
     "CppMatchingEngine",
@@ -99,6 +107,7 @@
     "Side",
     "TimeInForce",
     "ValuationMark",
+    "advance_until_fill_boundary",
     "build_equity_curve",
     "cpp_execution_engine_available",
     "default_execution_engine_factory",

src/ordersim/replay/__init__.py

@@ -1,6 +1,14 @@
 """Replay orchestration for strategy functions."""
 
+from ordersim.replay.boundary import BoundaryAdvance, advance_until_fill_boundary
 from ordersim.replay.compiled_events import CompiledEventColumns
 from ordersim.replay.simulator import Replay, ReplayGateway, ReplayResult
 
-__all__ = ["CompiledEventColumns", "Replay", "ReplayGateway", "ReplayResult"]
+__all__ = [
+    "BoundaryAdvance",
+    "CompiledEventColumns",
+    "Replay",
+    "ReplayGateway",
+    "ReplayResult",
+    "advance_until_fill_boundary",
+]

src/ordersim/replay/boundary.py

@@ -0,0 +1,70 @@
+"""Helpers for advancing replay state to execution boundaries."""
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from ordersim.replay.compiled_events import CompiledEventColumns
+from ordersim.sim import ExecutionEngine
+from ordersim.types import Fill, MBOEvent
+
+
+@dataclass(frozen=True, slots=True)
+class BoundaryAdvance:
+    """Result from advancing until the first passive fill or slice end."""
+
+    events_consumed: int
+    fills: tuple[Fill, ...]
+
+    @property
+    def stopped_on_fill(self) -> bool:
+        """Whether advancement stopped because a passive fill appeared."""
+
+        return bool(self.fills)
+
+
+def advance_until_fill_boundary(
+    engine: ExecutionEngine,
+    events: Sequence[MBOEvent],
+    *,
+    start: int = 0,
+    stop: int | None = None,
+    compiled_events: CompiledEventColumns | None = None,
+) -> BoundaryAdvance:
+    """Advance an engine until a passive fill appears or the slice ends.
+
+    This is the small bridge between scalar replay and compiled replay. If the
+    engine exposes a compiled boundary method and compiled columns are provided,
+    the engine advances through the slice internally. Otherwise this helper
+    falls back to the readable scalar path.
+    """
+
+    end = len(events) if stop is None else stop
+    _validate_slice(start=start, stop=end, length=len(events))
+    if start == end:
+        return BoundaryAdvance(events_consumed=0, fills=())
+
+    apply_compiled = getattr(engine, "apply_events_until_fill", None)
+    if compiled_events is not None and apply_compiled is not None:
+        events_consumed, fills = apply_compiled(compiled_events.slice(start, end))
+        return BoundaryAdvance(
+            events_consumed=events_consumed,
+            fills=tuple(fills),
+        )
+
+    fills: list[Fill] = []
+    for offset, event in enumerate(events[start:end], start=1):
+        event_fills = engine.apply_event(event)
+        if event_fills:
+            fills.extend(event_fills)
+            return BoundaryAdvance(events_consumed=offset, fills=tuple(fills))
+
+    return BoundaryAdvance(events_consumed=end - start, fills=())
+
+
+def _validate_slice(*, start: int, stop: int, length: int) -> None:
+    if start < 0:
+        raise ValueError("start must be non-negative")
+    if stop < start:
+        raise ValueError("stop must be greater than or equal to start")
+    if stop > length:
+        raise ValueError("stop cannot exceed the number of events")

tests/test_boundary_advance.py

@@ -0,0 +1,153 @@
+from decimal import Decimal
+
+import pytest
+
+from ordersim import (
+    CompiledEventColumns,
+    CppMatchingEngine,
+    MatchingEngine,
+    MBOEvent,
+    advance_until_fill_boundary,
+    cpp_execution_engine_available,
+)
+
+
+def boundary_events() -> tuple[MBOEvent, ...]:
+    return (
+        MBOEvent(
+            ts_ns=1,
+            action="add",
+            side="bid",
+            price=Decimal("100.0"),
+            size=1,
+            order_id=1,
+        ),
+        MBOEvent(
+            ts_ns=2,
+            action="trade",
+            side="bid",
+            price=Decimal("100.0"),
+            size=2,
+            order_id=2,
+        ),
+        MBOEvent(
+            ts_ns=3,
+            action="add",
+            side="ask",
+            price=Decimal("101.0"),
+            size=1,
+            order_id=3,
+        ),
+    )
+
+
+def test_boundary_advance_scalar_path_stops_on_first_passive_fill() -> None:
+    engine = MatchingEngine()
+    resting = engine.place_limit(side="buy", price=Decimal("100.0"), size=1)
+
+    advance = advance_until_fill_boundary(engine, boundary_events())
+
+    assert resting.order_id is not None
+    assert advance.events_consumed == 2
+    assert advance.stopped_on_fill is True
+    assert [(fill.order_id, fill.ts_ns, fill.size) for fill in advance.fills] == [
+        (resting.order_id, 2, 1),
+    ]
+    assert engine.book_top() == (None, None)
+
+
+@pytest.mark.skipif(
+    not cpp_execution_engine_available(),
+    reason="optional C++ execution engine is not built",
+)
+def test_boundary_advance_compiled_path_matches_scalar_boundary() -> None:
+    events = boundary_events()
+    columns = CompiledEventColumns.from_events(events, tick_size=Decimal("0.10"))
+    scalar_engine = MatchingEngine()
+    compiled_engine = CppMatchingEngine(tick_size=Decimal("0.10"))
+
+    scalar_order = scalar_engine.place_limit(
+        side="buy",
+        price=Decimal("100.0"),
+        size=1,
+    )
+    compiled_order = compiled_engine.place_limit(
+        side="buy",
+        price=Decimal("100.0"),
+        size=1,
+    )
+
+    scalar_advance = advance_until_fill_boundary(scalar_engine, events)
+    compiled_advance = advance_until_fill_boundary(
+        compiled_engine,
+        events,
+        compiled_events=columns,
+    )
+
+    assert scalar_order.order_id is not None
+    assert compiled_order.order_id is not None
+    assert scalar_advance.events_consumed == compiled_advance.events_consumed
+    assert scalar_advance.stopped_on_fill is True
+    assert compiled_advance.stopped_on_fill is True
+    assert [
+        (fill.side, fill.price, fill.size, fill.ts_ns)
+        for fill in scalar_advance.fills
+    ] == [
+        (fill.side, fill.price, fill.size, fill.ts_ns)
+        for fill in compiled_advance.fills
+    ]
+    assert scalar_engine.book_top() == compiled_engine.book_top()
+
+
+def test_boundary_advance_reports_slice_end_without_fill() -> None:
+    events = (
+        MBOEvent(
+            ts_ns=1,
+            action="add",
+            side="bid",
+            price=Decimal("100.0"),
+            size=1,
+            order_id=1,
+        ),
+        MBOEvent(
+            ts_ns=2,
+            action="add",
+            side="ask",
+            price=Decimal("101.0"),
+            size=1,
+            order_id=2,
+        ),
+    )
+    engine = MatchingEngine()
+
+    advance = advance_until_fill_boundary(engine, events, start=0, stop=2)
+
+    assert advance.events_consumed == 2
+    assert advance.stopped_on_fill is False
+    assert advance.fills == ()
+    assert engine.book_top() == (Decimal("100.0"), Decimal("101.0"))
+
+
+def test_boundary_advance_accepts_empty_slice() -> None:
+    events = boundary_events()
+    engine = MatchingEngine()
+
+    advance = advance_until_fill_boundary(engine, events, start=1, stop=1)
+
+    assert advance.events_consumed == 0
+    assert advance.stopped_on_fill is False
+    assert advance.fills == ()
+
+
+def test_boundary_advance_validates_slice_bounds() -> None:
+    events = boundary_events()
+    engine = MatchingEngine()
+
+    with pytest.raises(ValueError, match="start"):
+        advance_until_fill_boundary(engine, events, start=-1)
+
+    with pytest.raises(ValueError, match="greater than or equal"):
+        advance_until_fill_boundary(engine, events, start=2, stop=1)
+
+    with pytest.raises(ValueError, match="number of events"):
+        advance_until_fill_boundary(engine, events, stop=len(events) + 1)