Fix deterministic ordering for multi-instrument market shards

src/backtest_engine/market_data.py

@@ -54,6 +54,7 @@ class MarketDataValidationError(ValueError):
 
 DEFAULT_BATCH_SIZE = 65_536
 _HASH_CHUNK_SIZE = 1024 * 1024
+_OrderState = tuple[tuple[str, int], tuple[int, str], bool, bool]
 
 
 def _utc_timestamp(value: object, label: str) -> datetime:
@@ -129,8 +130,11 @@ def _verify_schema(schema: pa.Schema, policy: ExecutionPolicy) -> None:
     def _verify_batch_rows(
         batch: pa.RecordBatch,
         policy: ExecutionPolicy,
-        previous_timestamp_micros: int | None,
-    ) -> int | None:
+        order_state: _OrderState | None,
+    ) -> _OrderState | None:
+        instrument_values = batch.column(
+            batch.schema.get_field_index("instrument_id")
+        ).to_pylist()
         timestamp_values = batch.column(
             batch.schema.get_field_index("bar_start_at")
         ).cast(pa.int64()).to_pylist()
@@ -140,14 +144,32 @@ def _verify_batch_rows(
         period_start_micros = int(policy.period_start.timestamp() * 1_000_000)
         period_end_micros = int(policy.period_end.timestamp() * 1_000_000)
         zone = ZoneInfo(policy.timezone)
-        for timestamp_micros, session_date in zip(
-            timestamp_values, session_dates, strict=True
+        if order_state is None:
+            previous_instrument_key = None
+            previous_time_key = None
+            instrument_major = True
+            time_major = True
+        else:
+            (
+                previous_instrument_key,
+                previous_time_key,
+                instrument_major,
+                time_major,
+            ) = order_state
+        for instrument_id, timestamp_micros, session_date in zip(
+            instrument_values, timestamp_values, session_dates, strict=True
         ):
-            if (
-                previous_timestamp_micros is not None
-                and timestamp_micros < previous_timestamp_micros
-            ):
-                raise MarketDataValidationError("bar_start_at must be ordered")
+            instrument_key = (str(instrument_id), timestamp_micros)
+            time_key = (timestamp_micros, str(instrument_id))
+            if previous_instrument_key is not None and instrument_key <= previous_instrument_key:
+                instrument_major = False
+            if previous_time_key is not None and time_key <= previous_time_key:
+                time_major = False
+            if not instrument_major and not time_major:
+                raise MarketDataValidationError(
+                    "rows must be uniquely ordered by instrument_id, bar_start_at "
+                    "or by bar_start_at, instrument_id"
+                )
             if not period_start_micros <= timestamp_micros < period_end_micros:
                 raise MarketDataValidationError("bar_start_at is outside the pinned period")
             timestamp = datetime.fromtimestamp(
@@ -158,8 +180,16 @@ def _verify_batch_rows(
                 raise MarketDataValidationError(
                     "session_date_et does not match bar_start_at in policy timezone"
                 )
-            previous_timestamp_micros = timestamp_micros
-        return previous_timestamp_micros
+            previous_instrument_key = instrument_key
+            previous_time_key = time_key
+        if previous_instrument_key is None or previous_time_key is None:
+            return order_state
+        return (
+            previous_instrument_key,
+            previous_time_key,
+            instrument_major,
+            time_major,
+        )
 
     @staticmethod
     def _content_hash(path: Path) -> str:
@@ -253,19 +283,34 @@ def iter_batches(
         """Yield verified bounded batches without loading an object as one byte string.
 
         Hash verification is deliberately a streaming first pass. Parquet decoding is
-        then bounded by ``batch_size`` and validates ordering across row-group and
-        object boundaries before yielding each batch.
+        then bounded by ``batch_size``. Producer objects are canonically ordered
+        by either ``instrument_id, bar_start_at`` or ``bar_start_at, instrument_id``;
+        the event clock later forms the global time order without forcing this reader
+        to materialize the dataset.
         """
         self._validate_manifest(manifest, policy)
-        previous_timestamp_micros: int | None = None
+        schema: pa.Schema | None = None
+        parquets: list[pq.ParquetFile] = []
         for metadata in manifest["objects"]:
             parquet = self._parquet_file(metadata, policy)
+            object_schema = parquet.schema_arrow
+            if schema is None:
+                schema = object_schema
+            # Producers may attach shard-local provenance metadata. Each object
+            # already proves the required schema_version above; the logical
+            # Arrow fields, not unrelated file metadata, must match across the stream.
+            elif not schema.equals(object_schema, check_metadata=False):
+                raise MarketDataValidationError("Parquet object schemas do not match")
+            parquets.append(parquet)
+
+        if schema is None:  # pragma: no cover - both manifest contracts require objects
+            return
+
+        for parquet in parquets:
+            order_state: _OrderState | None = None
             try:
-                batches = parquet.iter_batches(batch_size=self._batch_size)
-                for batch in batches:
-                    previous_timestamp_micros = self._verify_batch_rows(
-                        batch, policy, previous_timestamp_micros
-                    )
+                for batch in parquet.iter_batches(batch_size=self._batch_size):
+                    order_state = self._verify_batch_rows(batch, policy, order_state)
                     yield batch
             except MarketDataValidationError:
                 raise

tests/test_market_data_reader.py

@@ -2,16 +2,19 @@
 
 import hashlib
 from copy import deepcopy
-from datetime import date
+from datetime import date, datetime, timezone
 from pathlib import Path
 
 import pyarrow as pa
 import pyarrow.parquet as pq
 import pytest
 
+from backtest_engine.calendar import XNYS_CALENDAR
 from backtest_engine.contracts import canonical_dataset_hash
+from backtest_engine.event_clock import MarketEventClock
 from backtest_engine.execution_policy import D17_EXECUTION_POLICY_FIXTURE
 from backtest_engine.market_data import MarketDataValidationError, ParquetMarketDataReader
+from backtest_engine.orchestrator import bar_events_from_batches
 from d_market_data_testkit import write_small_market_bars
 
 
@@ -113,6 +116,154 @@ def whole_file_read_is_forbidden(*_args: object, **_kwargs: object) -> object:
     ]
 
 
+def test_reader_accepts_eight_shard_instrument_major_data_and_clock_orders_events(
+    tmp_path: Path,
+) -> None:
+    """Pin the actual INT03 publication shape through the production event boundary.
+
+    Seven shards are empty and shard 04 contains two instrument-major series. Its
+    timestamps reset once at the instrument boundary, exactly as the producer's
+    canonical ``instrument_id, bar_start_at`` ordering requires. The event clock,
+    not the bounded Parquet reader, establishes global market-time order.
+    """
+    fixture = write_small_market_bars(tmp_path / "source.parquet")
+    source = pq.read_table(fixture.path)
+    objects: list[dict[str, object]] = []
+    for shard in range(8):
+        path = tmp_path / f"shard-{shard:02d}.parquet"
+        shard_table = source.slice(0, 0)
+        if shard == 4:
+            instruments = []
+            for instrument, symbol in (
+                ("11111111-1111-4111-8111-111111111111", "AAPL"),
+                ("22222222-2222-4222-8222-222222222222", "MSFT"),
+            ):
+                table = source.set_column(
+                    source.schema.get_field_index("instrument_id"),
+                    source.schema.field("instrument_id"),
+                    pa.array([instrument] * source.num_rows, type=pa.string()),
+                )
+                table = table.set_column(
+                    table.schema.get_field_index("provider_symbol"),
+                    table.schema.field("provider_symbol"),
+                    pa.array([symbol] * table.num_rows, type=pa.string()),
+                )
+                instruments.append(table)
+            shard_table = pa.concat_tables(instruments)
+        shard_table = shard_table.replace_schema_metadata(
+            {**(shard_table.schema.metadata or {}), b"shard_provenance": f"s{shard:02d}".encode()}
+        )
+        pq.write_table(shard_table, path, compression="zstd", version="2.6")
+        objects.append(
+            dict(
+                PINNED_OBJECT,
+                storage_object_id=f"33333333-3333-4333-8333-{shard + 1:012d}",
+                object_key=path.name,
+                content_hash=hashlib.sha256(path.read_bytes()).hexdigest(),
+                shard_key=f"s{shard:02d}-of-08",
+                row_count=shard_table.num_rows,
+            )
+        )
+    manifest = _manifest_for(tmp_path / "shard-00.parquet")
+    manifest["objects"] = objects
+    manifest["dataset_hash"] = canonical_dataset_hash(objects)
+
+    batches = list(
+        ParquetMarketDataReader(tmp_path, batch_size=3).iter_batches(
+            manifest,
+            D17_EXECUTION_POLICY_FIXTURE,
+        )
+    )
+    table = pa.Table.from_batches(batches)
+    events = bar_events_from_batches(batches, data_kind="BAR", resolution="30m")
+
+    assert table.num_rows == 4
+    assert list(
+        zip(
+            table["instrument_id"].to_pylist(),
+            table["bar_start_at"].cast(pa.int64()).to_pylist(),
+            strict=True,
+        )
+    ) == sorted(
+        zip(
+            table["instrument_id"].to_pylist(),
+            table["bar_start_at"].cast(pa.int64()).to_pylist(),
+            strict=True,
+        )
+    )
+    assert table["provider_symbol"].to_pylist() == ["AAPL", "AAPL", "MSFT", "MSFT"]
+    assert len(events) == 4
+    schedule = XNYS_CALENDAR.session_schedule(date(2024, 1, 1), date(2024, 3, 31))
+    released = MarketEventClock(schedule, events).advance_to(
+        datetime(2024, 1, 3, tzinfo=timezone.utc)
+    ).released_events
+    assert [event.occurred_at for event in released] == sorted(
+        event.occurred_at for event in released
+    )
+
+
+def test_reader_still_rejects_rows_out_of_order_inside_one_shard(tmp_path: Path) -> None:
+    fixture = write_small_market_bars(tmp_path / "market-bars.parquet")
+    table = pq.read_table(fixture.path).take(pa.array([1, 0]))
+    pq.write_table(table, fixture.path, compression="zstd", version="2.6")
+    manifest = _manifest_for(fixture.path)
+
+    with pytest.raises(MarketDataValidationError, match="uniquely ordered"):
+        ParquetMarketDataReader(tmp_path).read(manifest, D17_EXECUTION_POLICY_FIXTURE)
+
+
+def test_reader_rejects_duplicate_instrument_timestamp_key(tmp_path: Path) -> None:
+    fixture = write_small_market_bars(tmp_path / "market-bars.parquet")
+    table = pq.read_table(fixture.path).take(pa.array([0, 0]))
+    pq.write_table(table, fixture.path, compression="zstd", version="2.6")
+    manifest = _manifest_for(fixture.path)
+
+    with pytest.raises(MarketDataValidationError, match="uniquely ordered"):
+        ParquetMarketDataReader(tmp_path).read(manifest, D17_EXECUTION_POLICY_FIXTURE)
+
+
+def test_reader_rejects_instrument_order_regression(tmp_path: Path) -> None:
+    fixture = write_small_market_bars(tmp_path / "market-bars.parquet")
+    source = pq.read_table(fixture.path)
+    later_instrument = source.set_column(
+        source.schema.get_field_index("instrument_id"),
+        source.schema.field("instrument_id"),
+        pa.array(["22222222-2222-4222-8222-222222222222"] * source.num_rows),
+    )
+    table = pa.concat_tables([later_instrument, source]).take(pa.array([0, 2]))
+    pq.write_table(table, fixture.path, compression="zstd", version="2.6")
+    manifest = _manifest_for(fixture.path)
+
+    with pytest.raises(MarketDataValidationError, match="uniquely ordered"):
+        ParquetMarketDataReader(tmp_path).read(manifest, D17_EXECUTION_POLICY_FIXTURE)
+
+
+def test_reader_verifies_every_object_before_yielding_any_rows(tmp_path: Path) -> None:
+    first = write_small_market_bars(tmp_path / "first.parquet")
+    second = write_small_market_bars(tmp_path / "second.parquet")
+    objects = [
+        dict(
+            PINNED_OBJECT,
+            storage_object_id=f"33333333-3333-4333-8333-{index:012d}",
+            object_key=fixture.path.name,
+            content_hash=hashlib.sha256(fixture.path.read_bytes()).hexdigest(),
+            shard_key=f"s{index - 1:02d}-of-02",
+        )
+        for index, fixture in enumerate((first, second), start=1)
+    ]
+    manifest = _manifest_for(first.path)
+    manifest["objects"] = objects
+    manifest["dataset_hash"] = canonical_dataset_hash(objects)
+    second.path.write_bytes(second.path.read_bytes() + b"tampered")
+
+    batches = ParquetMarketDataReader(tmp_path).iter_batches(
+        manifest,
+        D17_EXECUTION_POLICY_FIXTURE,
+    )
+    with pytest.raises(MarketDataValidationError, match="content_hash"):
+        next(batches)
+
+
 def test_reader_rejects_object_bytes_that_do_not_match_manifest(tmp_path: Path) -> None:
     fixture = write_small_market_bars(tmp_path / "market-bars.parquet")
     manifest = _manifest_for(fixture.path)