feat: continuous batch account replay (15, 19)

README.md

@@ -1,5 +1,7 @@
 # quant-execution
 
+研究可信度升级：接口、使用示例、验收及限制见 [11–20 使用说明](docs/CONTINUOUS_REPLAY.md)。
+
 Hong Kong daily cash-equity research is available through
 `quant_execution.hong_kong.HKDailyExecution`: explicit board lots, dated and
 individually rounded fees, same-day sales, T+2 proceeds restrictions and the

docs/CONTINUOUS_REPLAY.md

@@ -0,0 +1,17 @@
+# Continuous batch replay (15, 19)
+
+`DeterministicRunEngine.replay_segments(segments, seed=...)` validates each nonempty segment
+and strict chronological separation before account mutation, then replays the entire event
+sequence **once**. The ledger, broker, unsettled amounts, orders, fees and strategy/risk state
+therefore persist across segment boundaries. Events inside a segment use the normal replay
+validation/order contract. Do not split events sharing the same `available_at` across segments.
+
+This API describes one uninterrupted batch account. It does not serialize a checkpoint or
+resume a live engine after process failure. Feed the complete frozen event stream on retry.
+
+`tests/test_continuous_replay.py` compares A-share, futures and crypto golden fixtures
+event-by-event against one-pass artifacts and terminal ledger snapshots. The cost test holds
+orders/fills fixed and independently computes cash = initial cash − purchase notional − fees
+using Decimal. A higher commission must reduce terminal NAV. This invariant does not apply
+to different adaptive order paths. Existing corporate-action tests remain the split/dividend
+regression basis; no external engine is claimed as an oracle.

pyproject.toml

@@ -12,7 +12,7 @@ requires-python = ">=3.10"
 dependencies = [
     "pyarrow>=14.0",
     "jsonschema>=4.20",
-    "quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@271a65ee383158b3dd7570e19a3f2b4123f09f79",
+    "quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@ba136c2fa2eea121bfb2ad7887b536c3952586f7",
 ]
 
 [project.optional-dependencies]

requirements.lock

@@ -67,7 +67,7 @@ pytz==2026.3.post1
     # via pandas
 pyyaml==6.0.3
     # via quant-data-kit
-quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@271a65ee383158b3dd7570e19a3f2b4123f09f79
+quant-data-kit @ git+https://github.com/PureSaber/quant-data-kit.git@ba136c2fa2eea121bfb2ad7887b536c3952586f7
     # via quant-execution (pyproject.toml)
 referencing==0.37.0
     # via

src/quant_execution/engine.py

@@ -211,6 +211,27 @@ def __init__(
         self.stored_artifacts: StoredRunArtifacts | None = None
         self._active_sink: ArrowReplayArtifactSink | None = None
 
+    def replay_segments(self, segments: Iterable[Iterable[MarketEvent]], seed: int) -> RunResult:
+        """Replay chronological OOS segments in ONE account and strategy lifecycle.
+
+        Fold boundaries do not reset cash, lots, receivables, broker orders, risk
+        reservations, or strategy latches. This is a batch continuous replay, not
+        a persisted checkpoint/restart API. Segments may not overlap in time.
+        """
+        validated = []
+        previous = None
+        for segment in segments:
+            events = self._validated_events(segment)
+            if not events or previous is not None and events[0].available_at <= previous:
+                raise ValidationError(
+                    "continuous segments must be nonempty and strictly chronological"
+                )
+            previous = events[-1].available_at
+            validated.append(events)
+        if not validated:
+            raise ValidationError("at least one continuous segment required")
+        return self.replay(chain.from_iterable(validated), seed=seed)
+
     def replay(self, event_stream: Iterable[MarketEvent], seed: int) -> RunResult:
         if isinstance(seed, bool) or not isinstance(seed, int) or seed < 0:
             raise ValidationError("seed must be a non-negative integer")

tests/test_continuous_replay.py

@@ -0,0 +1,55 @@
+from dataclasses import replace
+from decimal import Decimal
+
+import pytest
+from quant_data_kit.exceptions import ValidationError
+from test_engine import engine_for, fp, scenario_a_share, scenario_crypto, scenario_future
+
+
+@pytest.mark.parametrize("factory", [scenario_a_share, scenario_crypto, scenario_future])
+def test_segmentation_preserves_open_orders_cash_fees_and_exact_ledger(factory):
+    one, events = factory()
+    expected = one.replay(events, seed=0)
+    many, events = factory()
+    ordered = sorted(events, key=lambda e: e.available_at)
+    actual = many.replay_segments([[event] for event in ordered], seed=0)
+    assert actual == expected
+    assert many.artifacts == one.artifacts
+    assert many.ledger.snapshot() == one.ledger.snapshot()
+    assert many.strategy.calls == len(events)
+
+
+def test_invalid_segment_order_fails_before_account_mutation():
+    engine, events = scenario_a_share()
+    before = engine.ledger.snapshot()
+    with pytest.raises(ValidationError, match="chronological"):
+        engine.replay_segments([[events[0]], [events[1]]], seed=0)
+    assert engine.ledger.snapshot() == before
+    with pytest.raises(ValidationError, match="at least one"):
+        engine.replay_segments([], seed=0)
+
+
+def test_fixed_order_higher_cost_differential_against_hand_cash_account():
+    low, events = scenario_a_share()
+    low.replay(events, seed=0)
+    high, events = scenario_a_share()
+    instrument = next(iter(high.ledger.instruments))
+    original = high.ledger.instruments[instrument]
+    changed = replace(original, metadata={**original.metadata, "commission_rate": "0.02"})
+    high = engine_for(
+        run_id=high.run_id,
+        registry={instrument: changed},
+        initial_cash={"CNY": fp("100000")},
+        base_currency="CNY",
+        strategy=high.strategy,
+    )
+    high.replay(events, seed=0)
+    assert len(high.artifacts.fills) == len(low.artifacts.fills) == 1
+    low_fill, high_fill = low.artifacts.fills[0], high.artifacts.fills[0]
+    assert low_fill.quantity == high_fill.quantity and low_fill.price == high_fill.price
+    for engine in (low, high):
+        fill = engine.artifacts.fills[0]
+        paid = sum((fee.amount.to_decimal() for fee in engine.artifacts.fees), Decimal(0))
+        expected = Decimal(100000) - fill.quantity.to_decimal() * fill.price.to_decimal() - paid
+        assert engine.ledger.cash_balance("CNY") == expected
+    assert high.ledger.snapshot().nav.to_decimal() < low.ledger.snapshot().nav.to_decimal()