Use compact valuation mark columns

benchmarks/engine_throughput.py

@@ -48,6 +48,15 @@ def run_batch(engine: CppMatchingEngine, columns: CompiledEventColumns) -> None:
     engine.apply_events_batch(columns.slice(0, len(columns.ts_ns)))
 
 
+def run_batch_with_marks(
+    engine: CppMatchingEngine,
+    columns: CompiledEventColumns,
+) -> None:
+    """Apply one compiled event slice and return compact valuation marks."""
+
+    engine.apply_events_batch_with_marks(columns.slice(0, len(columns.ts_ns)))
+
+
 def measure(
     path_name: str,
     runner: Callable[[], None],
@@ -150,16 +159,28 @@ def main() -> None:
         repeats=args.repeats,
         warmups=args.warmups,
     )
-    results.extend((cpp_scalar, cpp_batch))
+    cpp_batch_marks = measure(
+        "CppMatchingEngine batch+marks",
+        lambda: run_batch_with_marks(
+            CppMatchingEngine(tick_size=TICK_SIZE),
+            columns,
+        ),
+        event_count=len(events),
+        repeats=args.repeats,
+        warmups=args.warmups,
+    )
+    results.extend((cpp_scalar, cpp_batch, cpp_batch_marks))
 
     for result in results[1:]:
         print(format_result(result))
 
     python_eps = results[0].events_per_second
     scalar_speedup = cpp_scalar.events_per_second / python_eps
     batch_speedup = cpp_batch.events_per_second / python_eps
+    batch_marks_speedup = cpp_batch_marks.events_per_second / python_eps
     print(f"per-event C++ speedup vs Python  {scalar_speedup:>7.2f}x")
     print(f"batch C++ speedup vs Python      {batch_speedup:>7.2f}x")
+    print(f"batch+marks speedup vs Python    {batch_marks_speedup:>7.2f}x")
 
 
 if __name__ == "__main__":

cpp/matching_engine_cpp.cpp

@@ -17,12 +17,6 @@ struct BatchBuffers {
     const int64_t* order_id;
 };
 
-struct MarkRow {
-    int64_t ts_ns;
-    int64_t bid_ticks;
-    int64_t ask_ticks;
-};
-
 template <typename T>
 const T* checked_buffer(
     const py::buffer& buffer,
@@ -104,6 +98,16 @@ BatchBuffers read_batch_buffers(
     };
 }
 
+py::bytes int64_bytes(const std::vector<int64_t>& values) {
+    if (values.empty()) {
+        return py::bytes();
+    }
+    return py::bytes(
+        reinterpret_cast<const char*>(values.data()),
+        values.size() * sizeof(int64_t)
+    );
+}
+
 std::vector<FillRow> apply_events_batch(
     MatchingEngineCpp& engine,
     const py::buffer& ts_ns,
@@ -158,7 +162,8 @@ py::tuple apply_events_batch_with_marks(
     );
 
     std::vector<FillRow> fills;
-    std::vector<MarkRow> marks;
+    std::vector<int64_t> mark_ts_ns;
+    std::vector<int64_t> mark_mid_ticks_x2;
 
     for (py::ssize_t i = 0; i < batch.n; ++i) {
         const auto event_fills = engine.apply_event_code(
@@ -173,11 +178,16 @@ py::tuple apply_events_batch_with_marks(
 
         const auto [bid_ticks, ask_ticks] = engine.book_top();
         if (bid_ticks >= 0 && ask_ticks >= 0) {
-            marks.push_back(MarkRow{batch.ts_ns[i], bid_ticks, ask_ticks});
+            mark_ts_ns.push_back(batch.ts_ns[i]);
+            mark_mid_ticks_x2.push_back(bid_ticks + ask_ticks);
         }
     }
 
-    return py::make_tuple(fills, marks);
+    return py::make_tuple(
+        fills,
+        int64_bytes(mark_ts_ns),
+        int64_bytes(mark_mid_ticks_x2)
+    );
 }
 
 py::tuple apply_events_until_fill(
@@ -228,11 +238,6 @@ PYBIND11_MODULE(_matching_engine_cpp, module) {
         .def_readonly("size", &FillRow::size)
         .def_readonly("ts_ns", &FillRow::ts_ns);
 
-    py::class_<MarkRow>(module, "MarkRow")
-        .def_readonly("ts_ns", &MarkRow::ts_ns)
-        .def_readonly("bid_ticks", &MarkRow::bid_ticks)
-        .def_readonly("ask_ticks", &MarkRow::ask_ticks);
-
     py::class_<MatchingEngineCpp>(module, "MatchingEngineCpp")
         .def(py::init<>())
         .def("apply_event", &MatchingEngineCpp::apply_event)

docs/benchmarks.md

@@ -34,10 +34,13 @@ This benchmark measures event ingestion through the execution engine itself:
 | `MatchingEngine` scalar | `apply_event(MBOEvent)` on the Python reference engine |
 | `CppMatchingEngine` per-event | `apply_event(MBOEvent)` one event at a time for compatibility checks |
 | `CppMatchingEngine` batch | `apply_events_batch(...)` over precompiled primitive columns |
+| `CppMatchingEngine` batch+marks | `apply_events_batch_with_marks(...)`, returning fills plus compact valuation-mark columns |
 
 The per-event C++ number is a diagnostic baseline, not the intended fast path.
 The useful compiled path is batched: Python hands C++ a contiguous slice and C++
-advances internally.
+advances internally. The `batch+marks` path is the closest engine-level
+measurement to ordinary audited replay because replay needs valuation marks for
+the equity curve.
 
 The batch result excludes `CompiledEventColumns.from_events(...)` construction.
 That conversion is meant to happen once before repeated compiled-engine runs;
@@ -67,8 +70,10 @@ also performs the work that makes `ordersim` inspectable:
 
 When the default C++ engine is available, ordinary replay advances each requested
 time slice through compiled columns and returns both fills and valuation marks.
-The Python reference engine remains event-by-event because it is the readable
-behavioral model.
+Those valuation marks are carried back as compact integer columns
+(`ts_ns[]`, `mid_ticks_x2[]`) and converted to public `Decimal` prices only when
+the equity curve is built. The Python reference engine remains event-by-event
+because it is the readable behavioral model.
 
 ## Interpreting Results
 

docs/economics.md

@@ -75,6 +75,12 @@ Replay builds `equity_curve` from observed fills and midpoint valuation marks.
 A midpoint mark is recorded when both bid and ask are available after replay
 applies a book event or order action.
 
+The C++ replay path transports those marks as compact integer columns and keeps
+midpoints as `bid_ticks + ask_ticks` until equity construction. Public
+`EquityPoint` values still expose `Decimal` prices; the compact representation
+only avoids creating an intermediate Python `ValuationMark` object for every
+market-data event.
+
 Replay only marks times it actually advances through. Full-session intraday
 drawdown therefore requires the strategy or harness to advance through the
 session window being studied, or to call `build_equity_curve(...)` directly with

docs/execution-engines.md

@@ -90,6 +90,8 @@ exposes a compiled batch-ingest path. It accepts primitive columns derived from
 the same `MBOEvent` schema and returns passive fills without changing the public
 matching semantics. Ordinary `Replay(...)` uses the compiled batch path when it
 can also receive the valuation marks needed to build the default equity curve.
+Those marks are transported as compact timestamp and midpoint-tick columns, not
+one Python object per market-data event.
 `Replay(...)` precompiles its immutable event stream once and shares that column
 view with each strategy run, so future compiled replay paths do not need to
 rebuild primitive columns inside `run_many(...)`.

docs/schema.md

@@ -109,6 +109,16 @@ See `docs/economics.md` for the assumptions and explicit non-goals.
 | `ts_ns` | `int` | Mark timestamp as UTC Unix-epoch nanoseconds. |
 | `price` | `Decimal` | Price used for open-lot valuation. |
 
+`CompiledValuationMarks` is the compact internal form used by the C++ replay
+path. It stores mark timestamps and midpoint prices as primitive integer
+columns:
+
+| Field | Type | Meaning |
+|---|---|---|
+| `ts_ns` | `memoryview[int64]` | Mark timestamps as UTC Unix-epoch nanoseconds. |
+| `mid_ticks_x2` | `memoryview[int64]` | `bid_ticks + ask_ticks`, preserving half-tick midpoints exactly. |
+| `tick_size` | `Decimal` | Price multiplier used when public `Decimal` prices are built. |
+
 `EquityPoint` is one output row from a mark-to-market equity curve.
 
 | Field | Type | Meaning |

src/ordersim/__init__.py

@@ -11,6 +11,7 @@
     write_parquet,
 )
 from ordersim.economics import (
+    CompiledValuationMarks,
     EquityPoint,
     ExecutionSummary,
     PositionLot,
@@ -68,6 +69,7 @@
     "BookSide",
     "BoundaryAdvance",
     "CompiledEventColumns",
+    "CompiledValuationMarks",
     "ConstantLatency",
     "CppMatchingEngine",
     "CsvSource",

src/ordersim/economics.py

@@ -1,7 +1,9 @@
 """Execution economics computed directly from fills."""
 
+from collections.abc import Iterable
 from dataclasses import dataclass
 from decimal import Decimal
+from typing import TypeAlias
 
 from ordersim.specs import InstrumentSpec
 from ordersim.types import Fill, Price, Side
@@ -43,6 +45,38 @@ class ValuationMark:
     price: Price
 
 
+@dataclass(frozen=True, slots=True)
+class CompiledValuationMarks:
+    """Compact valuation marks stored as timestamp and midpoint tick columns.
+
+    `mid_ticks_x2` stores `bid_ticks + ask_ticks`, so half-tick midpoints stay
+    exact until the public `Decimal` equity curve is built.
+    """
+
+    ts_ns: memoryview
+    mid_ticks_x2: memoryview
+    tick_size: Decimal
+
+    @classmethod
+    def from_bytes(
+        cls,
+        *,
+        ts_ns: bytes,
+        mid_ticks_x2: bytes,
+        tick_size: Decimal,
+    ) -> "CompiledValuationMarks":
+        """Build compact marks from native int64 byte columns."""
+
+        timestamps = memoryview(ts_ns).cast("q")
+        mids = memoryview(mid_ticks_x2).cast("q")
+        if len(timestamps) != len(mids):
+            raise ValueError("valuation mark columns must have equal length")
+        return cls(ts_ns=timestamps, mid_ticks_x2=mids, tick_size=tick_size)
+
+    def __len__(self) -> int:
+        return len(self.ts_ns)
+
+
 @dataclass(frozen=True, slots=True)
 class EquityPoint:
     """One point on a mark-to-market equity curve."""
@@ -56,6 +90,13 @@ class EquityPoint:
     drawdown: Decimal
 
 
+ValuationMarkInput: TypeAlias = (
+    tuple[ValuationMark | CompiledValuationMarks, ...]
+    | list[ValuationMark | CompiledValuationMarks]
+    | CompiledValuationMarks
+)
+
+
 def summarize_fills(
     fills: tuple[Fill, ...] | list[Fill],
     instrument: InstrumentSpec,
@@ -102,7 +143,7 @@ def summarize_fills(
 
 def build_equity_curve(
     fills: tuple[Fill, ...] | list[Fill],
-    marks: tuple[ValuationMark, ...] | list[ValuationMark],
+    marks: ValuationMarkInput,
     instrument: InstrumentSpec,
 ) -> tuple[EquityPoint, ...]:
     """Build a mark-to-market equity curve from fills and valuation marks.
@@ -112,31 +153,96 @@ def build_equity_curve(
     """
 
     sorted_fills = tuple(sorted(fills, key=lambda fill: fill.ts_ns))
-    sorted_marks = tuple(sorted(marks, key=lambda mark: mark.ts_ns))
+    if isinstance(marks, CompiledValuationMarks):
+        return _build_equity_curve_from_compiled_marks(
+            sorted_fills,
+            marks,
+            instrument,
+        )
+
+    sorted_marks = tuple(sorted(_iter_mark_pairs(marks), key=lambda mark: mark[0]))
+    open_lots: list[PositionLot] = []
+    realized_pnl = Decimal("0")
+    commission = Decimal("0")
+    high_water_mark = Decimal("0")
+    points: list[EquityPoint] = []
+    fill_index = 0
+
+    for mark_ts_ns, mark_price in sorted_marks:
+        while (
+            fill_index < len(sorted_fills)
+            and sorted_fills[fill_index].ts_ns <= mark_ts_ns
+        ):
+            fill = sorted_fills[fill_index]
+            realized_pnl += _apply_fill_to_lots(open_lots, fill, instrument)
+            commission += instrument.commission_per_contract * fill.size
+            fill_index += 1
+
+        unrealized_pnl = _unrealized_pnl(open_lots, mark_price, instrument)
+        equity = realized_pnl + unrealized_pnl - commission
+        high_water_mark = max(high_water_mark, equity)
+        points.append(
+            EquityPoint(
+                ts_ns=mark_ts_ns,
+                mark_price=mark_price,
+                realized_pnl=realized_pnl,
+                unrealized_pnl=unrealized_pnl,
+                commission=commission,
+                equity=equity,
+                drawdown=high_water_mark - equity,
+            )
+        )
+
+    return tuple(points)
+
+
+def _iter_mark_pairs(
+    marks: ValuationMarkInput,
+) -> Iterable[tuple[int, Price]]:
+    for mark in marks:
+        if isinstance(mark, CompiledValuationMarks):
+            yield from _iter_compiled_mark_pairs(mark)
+        else:
+            yield mark.ts_ns, mark.price
+
+
+def _iter_compiled_mark_pairs(
+    marks: CompiledValuationMarks,
+) -> Iterable[tuple[int, Price]]:
+    for ts_ns, mid_ticks_x2 in zip(marks.ts_ns, marks.mid_ticks_x2, strict=True):
+        yield ts_ns, marks.tick_size * Decimal(mid_ticks_x2) / 2
+
+
+def _build_equity_curve_from_compiled_marks(
+    sorted_fills: tuple[Fill, ...],
+    marks: CompiledValuationMarks,
+    instrument: InstrumentSpec,
+) -> tuple[EquityPoint, ...]:
     open_lots: list[PositionLot] = []
     realized_pnl = Decimal("0")
     commission = Decimal("0")
     high_water_mark = Decimal("0")
     points: list[EquityPoint] = []
     fill_index = 0
 
-    for mark in sorted_marks:
+    for ts_ns, mid_ticks_x2 in zip(marks.ts_ns, marks.mid_ticks_x2, strict=True):
         while (
             fill_index < len(sorted_fills)
-            and sorted_fills[fill_index].ts_ns <= mark.ts_ns
+            and sorted_fills[fill_index].ts_ns <= ts_ns
         ):
             fill = sorted_fills[fill_index]
             realized_pnl += _apply_fill_to_lots(open_lots, fill, instrument)
             commission += instrument.commission_per_contract * fill.size
             fill_index += 1
 
-        unrealized_pnl = _unrealized_pnl(open_lots, mark.price, instrument)
+        mark_price = marks.tick_size * Decimal(mid_ticks_x2) / 2
+        unrealized_pnl = _unrealized_pnl(open_lots, mark_price, instrument)
         equity = realized_pnl + unrealized_pnl - commission
         high_water_mark = max(high_water_mark, equity)
         points.append(
             EquityPoint(
-                ts_ns=mark.ts_ns,
-                mark_price=mark.price,
+                ts_ns=ts_ns,
+                mark_price=mark_price,
                 realized_pnl=realized_pnl,
                 unrealized_pnl=unrealized_pnl,
                 commission=commission,