diff --git a/src/backtest_engine/market_data.py b/src/backtest_engine/market_data.py index ecbe9a6..765ca36 100644 --- a/src/backtest_engine/market_data.py +++ b/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 diff --git a/tests/test_market_data_reader.py b/tests/test_market_data_reader.py index 0644006..00c543c 100644 --- a/tests/test_market_data_reader.py +++ b/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)