From d7748d84ebe62b7744444ab83c393b1e271a7b64 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Wed, 26 Aug 2026 17:32:07 +0900 Subject: [PATCH 1/9] feat: execute partition-aware backtests --- src/backtest_engine/basic_runtime.py | 85 +++++----- src/backtest_engine/execution_model.py | 16 +- src/backtest_engine/legacy_market_data.py | 21 +++ src/backtest_engine/market_data.py | 20 +-- src/backtest_engine/orchestrator.py | 46 +++++- src/backtest_engine/request_dispatch.py | 2 + src/backtest_engine/wiring.py | 180 ++++++++++++++++------ tests/test_basic_runtime.py | 84 ++++++++++ tests/test_execution_model.py | 22 +++ tests/test_feature_outputs.py | 4 +- tests/test_legacy_market_data.py | 37 +++++ tests/test_orchestrator.py | 37 ++++- tests/test_request_dispatch.py | 7 +- tests/test_wiring.py | 31 ++++ 14 files changed, 477 insertions(+), 115 deletions(-) diff --git a/src/backtest_engine/basic_runtime.py b/src/backtest_engine/basic_runtime.py index a7355b1..d86cff9 100644 --- a/src/backtest_engine/basic_runtime.py +++ b/src/backtest_engine/basic_runtime.py @@ -334,6 +334,7 @@ class BasicPlanFlow: condition_steps: tuple[PlanStep, ...] = () terminal_step: PlanStep | None = None allocation: str = "" + reference_series: tuple[str, str] = ("ADJUSTED_BAR", "$DATASET") @dataclass(frozen=True, slots=True) @@ -585,11 +586,16 @@ def load( condition_steps, terminal_step = chains[0] required_features = self._required_features(root, catalog) - reference_series = self._require_declared_features( - tuple(step for chain, _ in chains for step in chain), - required_features, - catalog, + chain_references = tuple( + self._require_declared_features(condition_steps, required_features, catalog) + for condition_steps, _terminal in chains ) + # Resolution belongs to a flow/container. Combining every condition chain here + # used to reject an otherwise valid multi-container strategy as soon as two + # partitions or flows used different bar sizes. The plan-level field is kept + # only for the v1 compatibility surface; all v2 execution and requirement + # derivation uses each BasicPlanFlow.reference_series. + reference_series = chain_references[0] self._require_compatibility(root, required_features, runtime_schema_version) version = snapshot["immutableStrategyVersion"] @@ -615,7 +621,7 @@ def load( allocation=terminal_step.argument("allocation"), required_features=required_features, reference_series=reference_series, - flows=self._load_flows(snapshot, chains, per_container), + flows=self._load_flows(snapshot, chains, chain_references, per_container), catalog=catalog, ) @@ -871,6 +877,7 @@ def _require_compatibility( def _load_flows( snapshot: Mapping[str, Any], chains: Sequence[tuple[tuple[PlanStep, ...], PlanStep]], + chain_references: Sequence[tuple[str, str]], per_container: bool, ) -> tuple[BasicPlanFlow, ...]: """Flatten the snapshot's flows, giving each the chain that belongs to it. @@ -899,6 +906,7 @@ def _load_flows( "repeated instrument would be allocated twice" ) condition_steps, terminal_step = chains[position if per_container else 0] + flow_reference = chain_references[position if per_container else 0] position += 1 flows.append( BasicPlanFlow( @@ -910,6 +918,7 @@ def _load_flows( condition_steps=condition_steps, terminal_step=terminal_step, allocation=terminal_step.argument("allocation"), + reference_series=flow_reference, ) ) return tuple(flows) @@ -1051,7 +1060,7 @@ def _evaluate_instrument( side=flow.side, status=BasicDecisionStatus.CANDIDATE, trace=tuple(trace), - reference_price=_reference_price(plan, evaluation), + reference_price=_reference_price(plan, flow, evaluation), ) # -- emission -------------------------------------------------------- @@ -1136,8 +1145,12 @@ def _allocate_equally(decisions: list[BasicInstrumentDecision], side: str) -> tu ) -def _reference_price(plan: BasicCompiledPlan, evaluation: ElementEvaluation) -> Decimal: - data_kind, resolution = plan.reference_series +def _reference_price( + plan: BasicCompiledPlan, flow: BasicPlanFlow, evaluation: ElementEvaluation +) -> Decimal: + data_kind, resolution = ( + plan.reference_series if flow.reference_series[1] == "$DATASET" else flow.reference_series + ) series = evaluation.inputs.series_for(data_kind, resolution) completed = series.completed_through(evaluation.as_of) if series else () if not completed: # pragma: no cover - a candidate loaded a feature from it @@ -1168,24 +1181,12 @@ def derive_data_requirements( same plan always makes the same request of the data layer. """ by_id: dict[str, DataRequirement] = {} - raw_lookback: dict[tuple[str, str], int] = {} - for flow in plan.flows: - for step in flow.condition_steps: - resolution = step.arguments.get("resolution") - if resolution is None: - continue - lookback = _raw_operation_lookback(step) - key = ("ADJUSTED_BAR", resolution) - raw_lookback[key] = max(raw_lookback.get(key, 1), lookback) - if not raw_lookback and plan.reference_series[1] != "$DATASET": - raw_lookback[plan.reference_series] = 1 - for instrument_id in plan.instrument_ids: - for feature in plan.required_features: + for feature in plan.required_features: + for instrument_id in feature.instruments: requirement_id = f"{instrument_id}|{feature.data_kind}|{feature.bar_resolution}" warmup_from = evaluation_from - feature.warmup_span existing = by_id.get(requirement_id) if existing is not None and existing.warmup_from <= warmup_from: - # Two features on the same series: the longer warm-up wins. continue by_id[requirement_id] = DataRequirement( requirement_id=requirement_id, @@ -1196,21 +1197,33 @@ def derive_data_requirements( evaluation_from=evaluation_from, evaluation_through=evaluation_through, ) - for (data_kind, resolution), bars in raw_lookback.items(): - requirement_id = f"{instrument_id}|{data_kind}|{resolution}" - warmup_from = evaluation_from - resolution_period(resolution) * bars - existing = by_id.get(requirement_id) - if existing is not None and existing.warmup_from <= warmup_from: + for flow in plan.flows: + raw_lookback: dict[tuple[str, str], int] = {} + for step in flow.condition_steps: + resolution = step.arguments.get("resolution") + if resolution is None: continue - by_id[requirement_id] = DataRequirement( - requirement_id=requirement_id, - instrument_id=instrument_id, - data_kind=data_kind, - resolution=resolution, - warmup_from=warmup_from, - evaluation_from=evaluation_from, - evaluation_through=evaluation_through, - ) + lookback = _raw_operation_lookback(step) + key = ("ADJUSTED_BAR", resolution) + raw_lookback[key] = max(raw_lookback.get(key, 1), lookback) + if not raw_lookback and flow.reference_series[1] != "$DATASET": + raw_lookback[flow.reference_series] = 1 + for instrument_id in flow.instrument_ids: + for (data_kind, resolution), bars in raw_lookback.items(): + requirement_id = f"{instrument_id}|{data_kind}|{resolution}" + warmup_from = evaluation_from - resolution_period(resolution) * bars + existing = by_id.get(requirement_id) + if existing is not None and existing.warmup_from <= warmup_from: + continue + by_id[requirement_id] = DataRequirement( + requirement_id=requirement_id, + instrument_id=instrument_id, + data_kind=data_kind, + resolution=resolution, + warmup_from=warmup_from, + evaluation_from=evaluation_from, + evaluation_through=evaluation_through, + ) return tuple(by_id[key] for key in sorted(by_id)) diff --git a/src/backtest_engine/execution_model.py b/src/backtest_engine/execution_model.py index 1df087e..758d02c 100644 --- a/src/backtest_engine/execution_model.py +++ b/src/backtest_engine/execution_model.py @@ -907,10 +907,24 @@ def process_bars(self, bars: Iterable[ExecutionBar]) -> tuple[Fill, ...]: fills.extend(self.process_bar(bar)) return tuple(fills) + def advance_to_bar(self, bar: ExecutionBar) -> tuple[BacktestOrder, ...]: + """Advance for a bar, permitting only a still-overlapping coarser bar.""" + if self._now is None or bar.starts_at >= self._now: + return self.advance_time(bar.starts_at) + if bar.ends_at < self._now: + raise ExecutionModelValidationError( + "a completed bar must not precede the execution clock" + ) + return () + def process_bar(self, bar: ExecutionBar) -> tuple[Fill, ...]: if not isinstance(bar, ExecutionBar): raise ExecutionModelValidationError("bar must be an ExecutionBar") - self.advance_time(bar.starts_at) + self.advance_to_bar(bar) + # Different resolutions overlap by definition. A 4h bar becomes visible + # after several 1h bars whose starts already advanced the shared account + # clock. Processing that still-open interval is not clock reversal: only + # orders eligible at the 4h bar's own start may fill below. if not bar.complete: return () # VOLUME_PARTICIPATION_RULE: one capacity per bar, shared by every order. diff --git a/src/backtest_engine/legacy_market_data.py b/src/backtest_engine/legacy_market_data.py index 49aed4a..7cb43d2 100644 --- a/src/backtest_engine/legacy_market_data.py +++ b/src/backtest_engine/legacy_market_data.py @@ -28,6 +28,7 @@ "is_legacy_market_loader_manifest", "legacy_dataset_hash", "legacy_period_matches", + "legacy_period_within_policy", "validate_legacy_market_loader_manifest", ] @@ -305,3 +306,23 @@ def legacy_period_matches( and _period_date(manifest.get("period_start"), "period_start") == local_start.date() and _period_date(manifest.get("period_end"), "period_end") == local_end.date() ) + + +def legacy_period_within_policy( + manifest: Mapping[str, Any], + period_start: datetime, + period_end: datetime, + timezone_name: str, +) -> bool: + """Accept one loader date-labelled segment contained by policy-local dates.""" + zone = ZoneInfo(timezone_name) + local_start = period_start.astimezone(zone) + local_end = period_end.astimezone(zone) + midnight = (0, 0, 0, 0) + manifest_start = _period_date(manifest.get("period_start"), "period_start") + manifest_end = _period_date(manifest.get("period_end"), "period_end") + return ( + (local_start.hour, local_start.minute, local_start.second, local_start.microsecond) == midnight + and (local_end.hour, local_end.minute, local_end.second, local_end.microsecond) == midnight + and local_start.date() <= manifest_start < manifest_end <= local_end.date() + ) diff --git a/src/backtest_engine/market_data.py b/src/backtest_engine/market_data.py index 67a12df..3ce1ed9 100644 --- a/src/backtest_engine/market_data.py +++ b/src/backtest_engine/market_data.py @@ -36,7 +36,7 @@ from .legacy_market_data import ( LEGACY_MARKET_SCHEMA_ID, is_legacy_market_loader_manifest, - legacy_period_matches, + legacy_period_within_policy, validate_legacy_market_loader_manifest, ) from .object_store.paths import long_path @@ -318,24 +318,18 @@ def _validate_manifest( if manifest.get("schema_id") != policy.market_data_schema_version: raise MarketDataValidationError("manifest schema_id does not match policy") if legacy: - if not legacy_period_matches( + if not legacy_period_within_policy( manifest, policy.period_start, policy.period_end, policy.timezone, ): - raise MarketDataValidationError("legacy manifest period does not match policy") + raise MarketDataValidationError("legacy manifest period is outside policy") else: - if ( - _utc_timestamp(manifest.get("period_start"), "manifest.period_start") - != policy.period_start - ): - raise MarketDataValidationError("manifest period_start does not match policy") - if ( - _utc_timestamp(manifest.get("period_end"), "manifest.period_end") - != policy.period_end - ): - raise MarketDataValidationError("manifest period_end does not match policy") + manifest_start = _utc_timestamp(manifest.get("period_start"), "manifest.period_start") + manifest_end = _utc_timestamp(manifest.get("period_end"), "manifest.period_end") + if not policy.period_start <= manifest_start < manifest_end <= policy.period_end: + raise MarketDataValidationError("manifest period is outside policy") def iter_batches( self, diff --git a/src/backtest_engine/orchestrator.py b/src/backtest_engine/orchestrator.py index 366fa27..7c2706f 100644 --- a/src/backtest_engine/orchestrator.py +++ b/src/backtest_engine/orchestrator.py @@ -421,6 +421,8 @@ class BacktestJob: resolution: str initial_cash: Decimal manifests: tuple[Mapping[str, Any], ...] = () + evaluation_from: datetime | None = None + evaluation_through: datetime | None = None def __post_init__(self) -> None: if not self.run_id: @@ -437,6 +439,13 @@ def __post_init__(self) -> None: if any(not isinstance(item, Mapping) for item in manifests): raise OrchestratorError("manifests must contain dataset manifest mappings") object.__setattr__(self, "manifests", manifests) + if (self.evaluation_from is None) != (self.evaluation_through is None): + raise OrchestratorError("evaluation interval must include both boundaries") + if self.evaluation_from is not None: + if self.evaluation_from.tzinfo is None or self.evaluation_through.tzinfo is None: + raise OrchestratorError("evaluation interval must be timezone-aware") + if self.evaluation_from >= self.evaluation_through: + raise OrchestratorError("evaluation interval must not be empty") # One source of truth for the bar period: the element catalog's # resolution table. A separately configured interval could disagree # with the resolution the plan actually reads. @@ -729,9 +738,14 @@ def run( monitor: ResourceMonitor, ) -> ReplayOutcome: schedule = self._schedule(job.execution_policy) - required_instrument_ids = frozenset( - requirement.instrument_id for requirement in job.requirements - ) + required_instruments_by_resolution: dict[str, frozenset[str]] = { + resolution: frozenset( + requirement.instrument_id + for requirement in job.requirements + if requirement.resolution == resolution + ) + for resolution in {requirement.resolution for requirement in job.requirements} + } try: combined_events: list[MarketDataEvent] = [] for manifest in job.manifests: @@ -742,6 +756,11 @@ def run( raise MarketDataValidationError( "every pinned dataset manifest must declare its resolution" ) + required_instrument_ids = required_instruments_by_resolution.get(resolution) + if not required_instrument_ids: + raise MarketDataValidationError( + f"pinned dataset resolution {resolution} is not required by the plan" + ) manifest_events = bar_events_from_batches( self._reader.iter_batches( manifest, @@ -758,9 +777,23 @@ def run( if len(job.manifests) > 1 else event for event in manifest_events ) + combined_events.sort(key=lambda event: ( + event.occurred_at, + resolution_period(str(event.payload.get("resolution", ""))), + event.instrument_id, + event.event_id, + )) events = tuple( replace(event, source_sequence=index) - for index, event in enumerate(combined_events, start=1) + for index, event in enumerate( + ( + event + for event in combined_events + if event.occurred_at >= min(requirement.warmup_from for requirement in job.requirements) + and (job.evaluation_through is None or event.occurred_at < job.evaluation_through) + ), + start=1, + ) ) except MarketDataValidationError: return self._abort( @@ -833,6 +866,11 @@ def _execute( per_series_limit=getattr(replay, "visible_event_limit", None), ) for sequence, instant in enumerate(sorted(events_at), start=1): + if job.evaluation_through is not None and instant >= job.evaluation_through: + break + if job.evaluation_from is not None and instant < job.evaluation_from: + visible_window.advance_to(instant) + continue try: lease = coordinator.heartbeat( lease, self._wall_clock(), monitor.sample() diff --git a/src/backtest_engine/request_dispatch.py b/src/backtest_engine/request_dispatch.py index aca4c43..54edcda 100644 --- a/src/backtest_engine/request_dispatch.py +++ b/src/backtest_engine/request_dispatch.py @@ -274,6 +274,8 @@ def __call__(self, request: Mapping[str, Any], lane: RequestLane) -> None: "datasetManifestId": str(primary.dataset_manifest_id), "expectedDatasetHash": primary.locked_dataset_hash, "expectedSnapshotHash": str(request["expectedSnapshotHash"]), + "evaluationStart": run.evaluation_start.isoformat(), + "evaluationEnd": run.evaluation_end.isoformat(), "datasets": [_dataset_payload(item) for item in stored_datasets], "featureMaterializations": [_feature_payload(item) for item in stored_features], **period_identity, diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 814a87e..84c910f 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -58,7 +58,7 @@ from collections.abc import Callable, Iterator, Mapping, Sequence from contextlib import contextmanager from dataclasses import dataclass, replace -from datetime import date, datetime, timedelta, timezone +from datetime import date, datetime, time, timedelta, timezone from decimal import ROUND_FLOOR, ROUND_HALF_EVEN, Decimal, localcontext from typing import Any, Protocol, cast from zoneinfo import ZoneInfo @@ -120,7 +120,7 @@ ) from .legacy_market_data import ( is_legacy_market_loader_manifest, - legacy_period_matches, + legacy_period_within_policy, validate_legacy_market_loader_manifest, ) from .lifecycle import BacktestLifecycleService, PersistenceRunGateway, SqsBacktestJobQueue @@ -479,7 +479,7 @@ def place(self, candidate: Any) -> str | None: def settle(self, event: MarketDataEvent) -> int: before = self._model.position(event.instrument_id) bar = self._bar_of(event) - for expired in self._model.advance_time(bar.starts_at): + for expired in self._model.advance_to_bar(bar): self._records.append( order_result_record(self._run, expired, bar.starts_at, self._model.cash, self._positions()) ) @@ -1119,6 +1119,8 @@ class JobEnvelope: feature_materializations: tuple[FeatureMaterializationPin, ...] evaluation_period_id: uuid.UUID | None input_set_hash: str | None + evaluation_start: date | None = None + evaluation_end: date | None = None @classmethod def parse(cls, job: Mapping[str, Any]) -> JobEnvelope: @@ -1188,6 +1190,16 @@ def parse(cls, job: Mapping[str, Any]) -> JobEnvelope: if job.get("inputSetHash") is not None else None ), + evaluation_start=( + date.fromisoformat(str(job["evaluationStart"])) + if job.get("evaluationStart") is not None + else None + ), + evaluation_end=( + date.fromisoformat(str(job["evaluationEnd"])) + if job.get("evaluationEnd") is not None + else None + ), ) except (KeyError, TypeError, ValueError) as exc: raise JobNotSatisfiable( @@ -1567,6 +1579,40 @@ def dataset_coverage(manifest: Mapping[str, Any]) -> tuple[datetime, datetime]: return min(starts), max(ends) +def segmented_dataset_coverage( + manifests: Sequence[Mapping[str, Any]], +) -> tuple[datetime, datetime]: + """Validate and join adjacent immutable segments of one exact resolution.""" + if not manifests: + raise JobNotSatisfiable( + "no dataset manifest segments were pinned", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + resolutions = {str(item.get("resolution", "")) for item in manifests} + if "" in resolutions or len(resolutions) != 1: + raise JobNotSatisfiable( + "a segmented dataset cover must declare one exact resolution", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + windows = sorted(dataset_coverage(item) for item in manifests) + start, end = windows[0] + for next_start, next_end in windows[1:]: + if next_start < end: + raise JobNotSatisfiable( + f"pinned {next(iter(resolutions))} dataset segments overlap at " + f"{next_start.isoformat()}", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + if next_start > end: + raise JobNotSatisfiable( + f"pinned {next(iter(resolutions))} dataset segments leave a gap " + f"{end.isoformat()}..{next_start.isoformat()}", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + end = max(end, next_end) + return start, end + + def evaluation_window(manifest: Mapping[str, Any], plan: BasicCompiledPlan) -> tuple[datetime, datetime]: """Where warm-up ends and evaluation begins, for this plan on this dataset. @@ -1610,7 +1656,7 @@ def require_compatible_execution_window( if legacy: try: validate_legacy_market_loader_manifest(manifest) - period_matches = legacy_period_matches( + period_matches = legacy_period_within_policy( manifest, policy.period_start, policy.period_end, @@ -1620,11 +1666,13 @@ def require_compatible_execution_window( problems.append(f"legacy dataset manifest is invalid: {exc}") period_matches = False else: - period_matches = manifest_start == policy.period_start and manifest_end == policy.period_end + period_matches = ( + policy.period_start <= manifest_start < manifest_end <= policy.period_end + ) if not period_matches: problems.append( "dataset manifest period " - f"{manifest_start.isoformat()}..{manifest_end.isoformat()} does not match " + f"{manifest_start.isoformat()}..{manifest_end.isoformat()} is outside " f"execution policy {policy.version} period " f"{policy.period_start.isoformat()}..{policy.period_end.isoformat()}" ) @@ -1887,6 +1935,10 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: f"compiled plan {plan_checksum} is not resolvable", reason_code="REQUIRED_INPUT_UNAVAILABLE", ) + try: + plan = self._runtime.load(plan_document, compiled_plan_checksum=plan_checksum) + except BasicPlanCompatibilityError as exc: + raise JobNotSatisfiable(str(exc), reason_code=exc.failure.value) from exc resolved_manifests: list[tuple[DatasetPin, Mapping[str, Any]]] = [] for pin in envelope.datasets: resolved = self._manifests.by_id(pin.manifest_id) @@ -1898,36 +1950,28 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: f"dataset manifest {pin.manifest_id} is missing or changed", reason_code="REQUIRED_INPUT_UNAVAILABLE", ) + if ( + not str(resolved.get("resolution", "")) + and len(envelope.datasets) == 1 + and plan.reference_series[1] != "$DATASET" + ): + resolved = dict(resolved, resolution=plan.reference_series[1]) resolved_manifests.append((pin, resolved)) for _pin, resolved in resolved_manifests: require_compatible_execution_window(policy, resolved, self._calendar) - declared_resolutions = [ - str(resolved.get("resolution", "")) for _pin, resolved in resolved_manifests - ] - if len(resolved_manifests) > 1 and ( - any(not resolution for resolution in declared_resolutions) - or len(set(declared_resolutions)) != len(declared_resolutions) - ): - raise JobNotSatisfiable( - "multiple pinned market datasets must declare unique resolutions", - reason_code="REQUIRED_INPUT_UNAVAILABLE", - ) - coverage_windows = {dataset_coverage(resolved) for _pin, resolved in resolved_manifests} - if len(coverage_windows) != 1: - raise JobNotSatisfiable( - "all pinned market datasets must share one evaluation period", - reason_code="REQUIRED_INPUT_UNAVAILABLE", - ) - try: - plan = self._runtime.load(plan_document, compiled_plan_checksum=plan_checksum) - except BasicPlanCompatibilityError as exc: - # The plan is immutable and addressed by its checksum. A schema, - # integrity, or catalog incompatibility therefore cannot become - # valid when SQS redelivers the same message. Translate it at the - # binding boundary so the existing terminal-result path records - # the precise plan-load failure once instead of retrying a - # deterministic producer/consumer mismatch to exhaustion. - raise JobNotSatisfiable(str(exc), reason_code=exc.failure.value) from exc + manifests_by_resolution: dict[str, list[Mapping[str, Any]]] = {} + for _pin, resolved in resolved_manifests: + resolution = str(resolved.get("resolution", "")) + if not resolution: + raise JobNotSatisfiable( + "every pinned market dataset must declare its resolution", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + manifests_by_resolution.setdefault(resolution, []).append(resolved) + coverage_by_resolution = { + resolution: segmented_dataset_coverage(manifests) + for resolution, manifests in manifests_by_resolution.items() + } if plan.reference_series[1] == "$DATASET": primary = [item for item in resolved_manifests if item[0].manifest_id == envelope.dataset_manifest_id] if len(primary) != 1: @@ -1943,30 +1987,64 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: reason_code="REQUIRED_INPUT_UNAVAILABLE", ) plan = replace( - plan, reference_series=(plan.reference_series[0], dataset_resolution) + plan, + reference_series=(plan.reference_series[0], dataset_resolution), + flows=tuple( + replace(flow, reference_series=(flow.reference_series[0], dataset_resolution)) + if flow.reference_series[1] == "$DATASET" + else flow + for flow in plan.flows + ), ) else: reference_resolution = plan.reference_series[1] - reference_manifests = [ - resolved for _pin, resolved in resolved_manifests - if str(resolved.get("resolution", "")) == reference_resolution + reference_manifests = manifests_by_resolution.get(reference_resolution, []) + if not reference_manifests: + raise JobNotSatisfiable( + f"the plan reference resolution {reference_resolution} has no pinned dataset cover", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + primary = [ + resolved for pin, resolved in resolved_manifests + if pin.manifest_id == envelope.dataset_manifest_id + and str(resolved.get("resolution", "")) == reference_resolution ] - if ( - not reference_manifests - and len(resolved_manifests) == 1 - and not str(resolved_manifests[0][1].get("resolution", "")) - ): - # Legacy single-dataset fixtures and messages predate the - # manifest-level resolution field. Multiple datasets never - # receive this fallback because their binding must be explicit. - reference_manifests = [resolved_manifests[0][1]] - if len(reference_manifests) != 1: + manifest = primary[0] if len(primary) == 1 else sorted( + reference_manifests, key=lambda item: dataset_coverage(item) + )[0] + if envelope.evaluation_start is not None or envelope.evaluation_end is not None: + if envelope.evaluation_start is None or envelope.evaluation_end is None: + raise JobNotSatisfiable( + "the explicit evaluation interval must include both boundaries", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + policy_zone = ZoneInfo(policy.timezone) + evaluation_from = datetime.combine( + envelope.evaluation_start, time.min, tzinfo=policy_zone + ).astimezone(timezone.utc) + evaluation_through = datetime.combine( + envelope.evaluation_end + timedelta(days=1), time.min, tzinfo=policy_zone + ).astimezone(timezone.utc) + if evaluation_from >= evaluation_through: + raise JobNotSatisfiable( + "the explicit evaluation interval is empty or reversed", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + else: + reference_start, reference_end = coverage_by_resolution[plan.reference_series[1]] + warmup = max( + (feature.warmup_span for feature in plan.required_features), + default=timedelta(0), + ) + evaluation_from, evaluation_through = reference_start + warmup, reference_end + for resolution, (coverage_start, coverage_end) in coverage_by_resolution.items(): + if coverage_start > evaluation_from or coverage_end < evaluation_through: raise JobNotSatisfiable( - f"the plan reference resolution {reference_resolution} must match exactly one pinned dataset", + f"pinned {resolution} dataset cover {coverage_start.isoformat()}.." + f"{coverage_end.isoformat()} does not contain evaluation interval " + f"{evaluation_from.isoformat()}..{evaluation_through.isoformat()}", reason_code="REQUIRED_INPUT_UNAVAILABLE", ) - manifest = reference_manifests[0] - evaluation_from, evaluation_through = evaluation_window(manifest, plan) feature_series: tuple[PinnedFeatureSeries, ...] = () if self._feature_materializations is not None or envelope.feature_materializations: if self._feature_materializations is None or self._feature_object_reader is None: @@ -2021,6 +2099,8 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: resolution=resolution, initial_cash=plan.initial_cash, manifests=tuple(resolved for _pin, resolved in resolved_manifests), + evaluation_from=evaluation_from, + evaluation_through=evaluation_through, ) except OrchestratorError as exc: raise JobNotSatisfiable(str(exc), reason_code="REQUIRED_INPUT_UNAVAILABLE") from exc diff --git a/tests/test_basic_runtime.py b/tests/test_basic_runtime.py index 2d2ff3f..3c34063 100644 --- a/tests/test_basic_runtime.py +++ b/tests/test_basic_runtime.py @@ -10,6 +10,7 @@ import copy import json +from dataclasses import replace from datetime import date, datetime, timedelta from decimal import Decimal from fractions import Fraction @@ -1092,6 +1093,44 @@ def test_requirements_are_deduplicated_across_flows_and_partitions() -> None: ] +def test_requirements_do_not_cross_product_unrelated_flow_instruments_and_clocks() -> None: + loaded = _runtime().load(_two_container_document()) + buy, sell = loaded.flows + sell_conditions = tuple( + replace( + step, + arguments=dict(step.arguments, resolution="30m") + if "resolution" in step.arguments else step.arguments, + ) + for step in sell.condition_steps + ) + plan = replace( + loaded, + flows=( + buy, + replace( + sell, + instrument_ids=(SECOND,), + condition_steps=sell_conditions, + reference_series=("ADJUSTED_BAR", "30m"), + ), + ), + ) + + requirements = derive_data_requirements( + plan, + evaluation_from=_utc("2025-11-28T14:45:00Z"), + evaluation_through=_utc("2025-11-28T20:00:00Z"), + ) + + assert [(item.instrument_id, item.resolution) for item in requirements] == [ + (FIRST, "1m"), + (SECOND, "30m"), + ] + assert plan.flows[0].reference_series == ("ADJUSTED_BAR", "1m") + assert plan.flows[1].reference_series == ("ADJUSTED_BAR", "30m") + + # --------------------------------------------------------------------------- # Replay: clock and availability gate actually applied # --------------------------------------------------------------------------- @@ -1397,6 +1436,51 @@ def test_loads_one_container_per_side_from_a_version_two_plan() -> None: assert sell.condition_steps[-1].arguments["operator"] == "GT" +def test_loads_raw_market_containers_with_independent_resolutions() -> None: + document = _two_container_document() + document["elementCatalogVersion"] = "basic-elements:2026-08-25" + document["requiredFeatures"] = [] + flows = document["executionSnapshot"]["partitions"][0]["flows"] + for flow, instrument_id, resolution in zip( + flows, + (FIRST, SECOND), + ("1h", "4h"), + strict=True, + ): + flow["officialInstrumentIds"] = [instrument_id] + price_step = { + "sequence": 1, + "operation": "PRICE_COMPARE", + "arguments": { + "operator": "GT", + "reference": "PREVIOUS_CLOSE", + "resolution": resolution, + }, + } + terminal_step = flow["steps"][-1] + terminal_step["sequence"] = 2 + terminal_step["arguments"].update( + { + "executionMode": "주기마다", + "maxExecutions": "20", + "maxPositionPercent": "25", + "orderPercent": "10", + "timeInForce": "DAY", + "waitInterval": "5", + "waitMode": "N봉 이후", + } + ) + flow["steps"] = [price_step, terminal_step] + document["planChecksum"] = compute_compiled_plan_checksum(document) + + plan = _runtime().load(document) + + assert [flow.reference_series for flow in plan.flows] == [ + ("ADJUSTED_BAR", "1h"), + ("ADJUSTED_BAR", "4h"), + ] + + def test_a_version_two_plan_missing_per_flow_steps_is_refused() -> None: """The version says the steps are per container; there is no fallback.""" document = copy.deepcopy(_document()) diff --git a/tests/test_execution_model.py b/tests/test_execution_model.py index d13df98..47f8a1a 100644 --- a/tests/test_execution_model.py +++ b/tests/test_execution_model.py @@ -17,6 +17,7 @@ CHART_OF_ACCOUNTS_VERSION, D23_MICROSTRUCTURE_POLICY_V1, BacktestExecutionModel, + ExecutionBar, ExecutionMicrostructurePolicy, ExecutionModelValidationError, InstrumentFractionalPolicy, @@ -144,6 +145,27 @@ def _bar( ) +def test_overlapping_multi_resolution_bars_do_not_reverse_the_execution_clock() -> None: + model = _model() + one_hour = ExecutionBar( + instrument_id=INSTRUMENT, + starts_at=_utc("2025-11-28T17:00:00Z"), + ends_at=_utc("2025-11-28T18:00:00Z"), + open=Decimal("100"), high=Decimal("101"), low=Decimal("99"), + close=Decimal("100"), volume=Decimal("1000"), resolution="1h", + ) + four_hour = ExecutionBar( + instrument_id=OTHER_INSTRUMENT, + starts_at=_utc("2025-11-28T14:00:00Z"), + ends_at=_utc("2025-11-28T18:00:00Z"), + open=Decimal("200"), high=Decimal("202"), low=Decimal("198"), + close=Decimal("200"), volume=Decimal("1000"), resolution="4h", + ) + + assert model.process_bar(one_hour) == () + assert model.process_bar(four_hour) == () + + def _entry( *, entry_id: str = ENTRY_ID, diff --git a/tests/test_feature_outputs.py b/tests/test_feature_outputs.py index 9c912f5..bd0e314 100644 --- a/tests/test_feature_outputs.py +++ b/tests/test_feature_outputs.py @@ -789,8 +789,8 @@ def test_incompatible_development_windows_are_one_terminal_binding_failure( row_count=len(CLOSES), coverage_end=EVALUATION_THROUGH, ) - manifest["period_start"] = "2024-01-01T05:00:00Z" - manifest["period_end"] = "2024-02-01T05:00:00Z" + manifest["period_start"] = "2015-01-01T05:00:00Z" + manifest["period_end"] = "2015-02-01T05:00:00Z" handler = _handler(Source({}), Reader(b"")) handler._policies = ExecutionPolicyCatalog([policy]) handler._manifests = StaticDatasetManifestSource({DATASET_MANIFEST_ID: manifest}) diff --git a/tests/test_legacy_market_data.py b/tests/test_legacy_market_data.py index 5905036..1524b4e 100644 --- a/tests/test_legacy_market_data.py +++ b/tests/test_legacy_market_data.py @@ -52,6 +52,18 @@ def test_exact_development_manifest_binds_to_the_one_month_et_policy() -> None: require_compatible_execution_window(policy, _fixture(), XNYS_CALENDAR) +def test_legacy_manifest_may_be_one_segment_inside_a_longer_policy() -> None: + policy = replace( + D17_EXECUTION_POLICY_FIXTURE, + version="development-official-backtest-2026-q3-v3", + period_start=datetime(2024, 1, 1, 5, tzinfo=timezone.utc), + period_end=datetime(2025, 1, 1, 5, tzinfo=timezone.utc), + market_data_schema_version="market-bars/1", + ) + + require_compatible_execution_window(policy, _fixture(), XNYS_CALENDAR) + + @pytest.mark.parametrize( "mutate", [ @@ -143,6 +155,31 @@ def test_reader_consumes_legacy_parquet_without_weakening_the_canonical_path( assert result.num_rows == 2 +def test_reader_consumes_one_legacy_segment_inside_a_longer_policy(tmp_path: Path) -> None: + key_root = ( + tmp_path + / "historical/provider=alpaca/feed=sip/adjustment=all/session=regular" + / "resolution=30m/revision=00000001/year=2024/shard=00-of-01" + / "manifest_id=11111111-1111-4111-8111-111111111111" + ) + extended_key_root = Path(long_path(key_root)) + extended_key_root.mkdir(parents=True) + fixture = write_small_market_bars(extended_key_root / "part-00001.parquet") + table = pq.read_table(fixture.path).replace_schema_metadata( + {b"schema_version": b"market-bars/1", b"processing_version": b"market-loader/1.0.0"} + ) + pq.write_table(table, fixture.path, compression="zstd", version="2.6") + manifest = _one_shard_manifest(fixture.path) + policy = replace( + D17_EXECUTION_POLICY_FIXTURE, + period_start=datetime(2024, 1, 1, 5, tzinfo=timezone.utc), + period_end=datetime(2025, 1, 1, 5, tzinfo=timezone.utc), + market_data_schema_version="market-bars/1", + ) + + assert ParquetMarketDataReader(tmp_path).read(manifest, policy).num_rows == 2 + + def test_composite_legacy_manifest_hashes_and_validates_multiple_source_years() -> None: manifest = _one_shard_manifest(Path(FIXTURE)) composite_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index 2fee5de..28fb08e 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -18,6 +18,7 @@ import hashlib from collections.abc import Mapping +from dataclasses import replace from datetime import date, datetime, timedelta, timezone from decimal import Decimal from fractions import Fraction @@ -538,6 +539,24 @@ def _run( return outcome, harness, coordinator +def test_explicit_evaluation_interval_keeps_warmup_visible_but_never_trades_outside_it( + tmp_path: Path, +) -> None: + path = tmp_path / "bars.parquet" + write_bars(path) + job = replace( + _job(manifest_for(path)), + evaluation_from=SECOND, + evaluation_through=THIRD, + ) + + outcome, harness, _ = _run(tmp_path, job=job) + + assert outcome.status is ReplayStatus.COMPLETED + assert [step.instant for step in outcome.steps] == [SECOND] + assert all(candidate.decided_at == SECOND for candidate in harness.engine.placed) + + # -------------------------------------------------------------------------- # The seams line up with BT-a's real classes # -------------------------------------------------------------------------- @@ -600,10 +619,9 @@ def test_orchestrator_reads_every_pinned_resolution_into_one_replay_clock() -> N _bar_row(_utc(14, 45), date(2024, 1, 2))], schema=_SCHEMA, ) - rows_30m = pa.Table.from_pylist( - [_bar_row(_utc(14, 30), date(2024, 1, 2))], - schema=_SCHEMA, - ) + row_30m = _bar_row(_utc(14, 30), date(2024, 1, 2)) + row_30m.update(instrument_id=MSFT, provider_symbol="MSFT") + rows_30m = pa.Table.from_pylist([row_30m], schema=_SCHEMA) manifests = ({"resolution": "15m"}, {"resolution": "30m"}) class Reader: @@ -617,8 +635,10 @@ def iter_batches( *, instrument_ids: frozenset[str] | None = None, ) -> Any: - assert instrument_ids == frozenset({AAPL}) resolution = str(manifest["resolution"]) + assert instrument_ids == ( + frozenset({AAPL}) if resolution == "15m" else frozenset({MSFT}) + ) self.seen.append(resolution) return (rows_15m if resolution == "15m" else rows_30m).to_batches() @@ -640,7 +660,7 @@ def iter_batches( evaluation_through=_utc(15, 0), ), DataRequirement( - requirement_id="aapl-30m", instrument_id=AAPL, data_kind=DATA_KIND, + requirement_id="msft-30m", instrument_id=MSFT, data_kind=DATA_KIND, resolution="30m", warmup_from=_utc(14, 30), evaluation_from=_utc(14, 30), evaluation_through=_utc(15, 0), ), @@ -666,8 +686,9 @@ def iter_batches( assert outcome.status is ReplayStatus.COMPLETED assert reader.seen == ["15m", "30m"] - visible = runtime.inputs_by_instant[_utc(15, 0)][AAPL] - assert {series.resolution for series in visible.series} == {"15m", "30m"} + visible = runtime.inputs_by_instant[_utc(15, 0)] + assert {series.resolution for series in visible[AAPL].series} == {"15m"} + assert {series.resolution for series in visible[MSFT].series} == {"30m"} # -------------------------------------------------------------------------- diff --git a/tests/test_request_dispatch.py b/tests/test_request_dispatch.py index bc01c6a..00dfe10 100644 --- a/tests/test_request_dispatch.py +++ b/tests/test_request_dispatch.py @@ -116,7 +116,8 @@ def test_publishes_existing_provider_created_run_as_an_execution_job( ) -> None: request = factory() queue = Queue() - publisher = BacktestRequestJobPublisher(Source(projection(request, lane)), queue) + run = projection(request, lane) + publisher = BacktestRequestJobPublisher(Source(run), queue) publisher(request, lane) @@ -139,6 +140,8 @@ def test_publishes_existing_provider_created_run_as_an_execution_job( else request["periods"][0]["datasets"][0]["expectedDatasetHash"] ), "expectedSnapshotHash": request["expectedSnapshotHash"], + "evaluationStart": run.evaluation_start.isoformat(), + "evaluationEnd": run.evaluation_end.isoformat(), "datasets": [ { "datasetManifestId": str(DATASET_ID), @@ -273,6 +276,8 @@ def test_basic_request_is_dispatched_through_the_same_pinned_two_stage_boundary( "datasetManifestId": request["datasetManifestId"], "expectedDatasetHash": request["expectedDatasetHash"], "expectedSnapshotHash": request["expectedSnapshotHash"], + "evaluationStart": run.evaluation_start.isoformat(), + "evaluationEnd": run.evaluation_end.isoformat(), "datasets": [ { "datasetManifestId": request["datasetManifestId"], diff --git a/tests/test_wiring.py b/tests/test_wiring.py index 7559b96..9516f6a 100644 --- a/tests/test_wiring.py +++ b/tests/test_wiring.py @@ -70,6 +70,7 @@ _metric_percent, dataset_coverage, evaluation_window, + segmented_dataset_coverage, ) from backtest_engine.worker import JobContext, JobResult from d_reproducibility_testkit import ( @@ -663,6 +664,36 @@ def test_dataset_coverage_reads_the_objects_not_the_dataset_window() -> None: assert dataset_coverage(widened)[1] == FIRST_BAR_START + BAR * len(CLOSES) +def test_adjacent_manifests_of_the_same_resolution_form_one_cover() -> None: + first = dataset_manifest("1" * 64, row_count=1, coverage_end=FIRST_BAR_START + BAR) + second = dataset_manifest("2" * 64, row_count=1, coverage_end=FIRST_BAR_START + BAR) + for manifest in (first, second): + manifest["resolution"] = "1h" + first["objects"][0]["period_start"] = "2024-01-01T00:00:00Z" + first["objects"][0]["period_end"] = "2025-01-01T00:00:00Z" + second["objects"][0]["period_start"] = "2025-01-01T00:00:00Z" + second["objects"][0]["period_end"] = "2026-01-01T00:00:00Z" + + assert segmented_dataset_coverage((second, first)) == ( + datetime(2024, 1, 1, tzinfo=UTC), + datetime(2026, 1, 1, tzinfo=UTC), + ) + + +def test_segmented_manifest_cover_rejects_a_gap() -> None: + first = dataset_manifest("1" * 64, row_count=1, coverage_end=FIRST_BAR_START + BAR) + second = dataset_manifest("2" * 64, row_count=1, coverage_end=FIRST_BAR_START + BAR) + for manifest in (first, second): + manifest["resolution"] = "4h" + first["objects"][0]["period_start"] = "2024-01-01T00:00:00Z" + first["objects"][0]["period_end"] = "2025-01-01T00:00:00Z" + second["objects"][0]["period_start"] = "2025-02-01T00:00:00Z" + second["objects"][0]["period_end"] = "2026-01-01T00:00:00Z" + + with pytest.raises(JobNotSatisfiable, match="gap"): + segmented_dataset_coverage((first, second)) + + def test_the_pinned_completion_instant_follows_every_replay_instant() -> None: """Guards the fixture: `completed_at` must not precede a result record.""" assert COMPLETED_AT > FIRST_BAR_START + BAR * len(CLOSES) From 5b7ca01277dc8457e973b0ecdd8713fdb2ac4e34 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Wed, 26 Aug 2026 17:37:43 +0900 Subject: [PATCH 2/9] fix: bind scoped and position-only inputs --- src/backtest_engine/production.py | 3 ++- src/backtest_engine/wiring.py | 39 +++++++++++++++++++++++++++---- 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/src/backtest_engine/production.py b/src/backtest_engine/production.py index 6b77a00..caff031 100644 --- a/src/backtest_engine/production.py +++ b/src/backtest_engine/production.py @@ -246,7 +246,7 @@ def __init__(self, engine: Engine) -> None: def by_id(self, manifest_id: uuid.UUID) -> Mapping[str, Any] | None: manifest_sql = text( """ - SELECT manifest.id, manifest.revision_number, manifest.status, + SELECT manifest.id, manifest.instrument_id, manifest.revision_number, manifest.status, manifest.dataset_hash, manifest.schema_version, manifest.period_start, manifest.period_end, manifest.available_at, provider.code AS provider_code, feed.code AS feed_code, @@ -338,6 +338,7 @@ def by_id(self, manifest_id: uuid.UUID) -> Mapping[str, Any] | None: "contract_id": "com06.dataset-manifest", "schema_version": 1, "manifest_id": str(row["id"]), + "instrument_id": str(row.get("instrument_id")) if row.get("instrument_id") else None, "dataset_id": _dataset_id( [item["object_key"] for item in objects], manifest_id, diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 84c910f..53a6843 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -1968,10 +1968,19 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: reason_code="REQUIRED_INPUT_UNAVAILABLE", ) manifests_by_resolution.setdefault(resolution, []).append(resolved) - coverage_by_resolution = { - resolution: segmented_dataset_coverage(manifests) - for resolution, manifests in manifests_by_resolution.items() - } + coverage_by_resolution: dict[str, tuple[datetime, datetime]] = {} + for resolution, manifests in manifests_by_resolution.items(): + manifests_by_scope: dict[str, list[Mapping[str, Any]]] = {} + for resolved in manifests: + manifests_by_scope.setdefault(str(resolved.get("instrument_id") or "*"), []).append(resolved) + scoped_windows = [ + segmented_dataset_coverage(scoped) + for scoped in manifests_by_scope.values() + ] + coverage_by_resolution[resolution] = ( + max(window[0] for window in scoped_windows), + min(window[1] for window in scoped_windows), + ) if plan.reference_series[1] == "$DATASET": primary = [item for item in resolved_manifests if item[0].manifest_id == envelope.dataset_manifest_id] if len(primary) != 1: @@ -2031,6 +2040,24 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: reason_code="REQUIRED_INPUT_UNAVAILABLE", ) else: + resolved_flows = [] + for flow in plan.flows: + if flow.reference_series[1] != "$DATASET": + resolved_flows.append(flow) + continue + inherited = { + candidate.reference_series + for candidate in plan.flows + if candidate.reference_series[1] != "$DATASET" + and set(candidate.instrument_ids).intersection(flow.instrument_ids) + } + if len(inherited) != 1: + raise JobNotSatisfiable( + f"position-only flow {flow.flow_id} has no unambiguous market clock", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + resolved_flows.append(replace(flow, reference_series=next(iter(inherited)))) + plan = replace(plan, flows=tuple(resolved_flows)) reference_start, reference_end = coverage_by_resolution[plan.reference_series[1]] warmup = max( (feature.warmup_span for feature in plan.required_features), @@ -2068,6 +2095,10 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: requirements = derive_data_requirements( plan, evaluation_from=evaluation_from, evaluation_through=evaluation_through ) + requirements = tuple( + replace(requirement, warmup_from=max(requirement.warmup_from, policy.period_start)) + for requirement in requirements + ) if not requirements: # pragma: no cover - a loaded plan always declares one raise JobNotSatisfiable( "the compiled plan declares no data requirement", From db0d30f2d1c02c1afa75326aba278fb90b28cb74 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Thu, 27 Aug 2026 12:53:12 +0900 Subject: [PATCH 3/9] feat(backtest): add owner soft-delete lifecycle --- ...090000__backtest_add_owner_soft_delete.sql | 24 ++++++ src/backtest_engine/api.py | 24 ++++++ src/backtest_engine/lifecycle.py | 78 ++++++++++++++++++- src/backtest_engine/orchestrator.py | 6 +- src/backtest_engine/persistence/protocols.py | 2 + .../persistence/repositories.py | 58 +++++++++++++- src/backtest_engine/persistence/rows.py | 2 + src/backtest_engine/persistence/tables.py | 3 + src/backtest_engine/wiring.py | 14 ++-- tests/persistence/test_roundtrip.py | 51 ++++++++++++ tests/test_backtest_api.py | 39 ++++++++++ tests/test_lifecycle.py | 47 +++++++++++ 12 files changed, 338 insertions(+), 10 deletions(-) create mode 100644 db/migration-contributions/migrations/V20260827090000__backtest_add_owner_soft_delete.sql diff --git a/db/migration-contributions/migrations/V20260827090000__backtest_add_owner_soft_delete.sql b/db/migration-contributions/migrations/V20260827090000__backtest_add_owner_soft_delete.sql new file mode 100644 index 0000000..eef1993 --- /dev/null +++ b/db/migration-contributions/migrations/V20260827090000__backtest_add_owner_soft_delete.sql @@ -0,0 +1,24 @@ +ALTER TABLE backtest.runs + ADD COLUMN deletion_requested_at timestamptz, + ADD COLUMN deleted_at timestamptz; + +ALTER TABLE backtest.runs + ADD CONSTRAINT backtest_deletion_state_consistent + CHECK ( + deleted_at IS NULL + OR (deletion_requested_at IS NOT NULL AND deleted_at >= deletion_requested_at) + ), + ADD CONSTRAINT backtest_deleted_run_is_terminal + CHECK ( + deleted_at IS NULL + OR status IN ('COMPLETED', 'FAILED', 'UNAVAILABLE', 'CANCELLED') + ); + +CREATE INDEX runs_owner_account_id_deleted_at_queued_at_idx + ON backtest.runs (owner_account_id, deleted_at, queued_at DESC); + +COMMENT ON COLUMN backtest.runs.deletion_requested_at IS + 'Owner-requested removal time. Running work remains durable until cooperative cancellation reaches a terminal state.'; + +COMMENT ON COLUMN backtest.runs.deleted_at IS + 'Soft-delete completion time. Deleted runs are hidden from owner APIs but retained as immutable execution evidence.'; diff --git a/src/backtest_engine/api.py b/src/backtest_engine/api.py index d926d67..1bff283 100644 --- a/src/backtest_engine/api.py +++ b/src/backtest_engine/api.py @@ -192,6 +192,8 @@ def _run_payload(run: BacktestRun) -> dict[str, Any]: "cancellationRequestedAt": _iso(row.cancellation_requested_at), "cancellationReasonCode": row.cancellation_reason_code, "cancelledAt": _iso(row.cancelled_at), + "deletionRequestedAt": _iso(row.deletion_requested_at), + "deletedAt": _iso(row.deleted_at), "attemptCount": len(run.attempts), } @@ -545,6 +547,28 @@ def cancel_backtest( raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc return {"run": _run_payload(run), "cancellationRequested": True} + @app.delete( + f"{API_PREFIX}/backtests/{{run_id}}", + status_code=status.HTTP_202_ACCEPTED, + tags=["backtests"], + ) + def delete_backtest(run_id: UUID, principal: Principal = Auth) -> dict[str, Any]: + """Owner delete with evidence retention and cooperative running cancellation.""" + try: + run = lifecycle.request_deletion( + run_id, + owner_account_id=principal.account_id, + ) + except BacktestRunNotFound as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + except NotRunOwner as exc: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc + return { + "run": _run_payload(run), + "deletionRequested": True, + "deleted": run.run.deleted_at is not None, + } + @app.get(f"{API_PREFIX}/backtests/{{run_id}}/attempts", tags=["backtests"]) def get_attempts(run_id: UUID, principal: Principal = Auth) -> dict[str, Any]: """Query 3/8: the durable attempt history behind a run.""" diff --git a/src/backtest_engine/lifecycle.py b/src/backtest_engine/lifecycle.py index fe82e34..6d75195 100644 --- a/src/backtest_engine/lifecycle.py +++ b/src/backtest_engine/lifecycle.py @@ -407,6 +407,8 @@ def request_cancellation( self, run_id: UUID, *, reason_code: str, requested_at: datetime ) -> RunRow: ... + def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: ... + class PersistenceRunGateway: """Durable `RunGateway` over the canonical schema, in SQLAlchemy Core.""" @@ -543,6 +545,13 @@ def request_cancellation( except RowNotFound as exc: raise BacktestRunNotFound(str(exc)) from exc + def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: + try: + with self._write() as uow: + return uow.runs.request_deletion(run_id, requested_at=requested_at) + except RowNotFound as exc: + raise BacktestRunNotFound(str(exc)) from exc + class InMemoryRunGateway: """Faithful in-process `RunGateway`, for tests about HTTP rather than SQL. @@ -629,7 +638,11 @@ def get(self, run_id: UUID) -> RunRow: def list_by_owner(self, owner_account_id: UUID, *, limit: int, offset: int) -> tuple[RunRow, ...]: with self._lock: - owned = [row for row in self._runs.values() if row.owner_account_id == owner_account_id] + owned = [ + row + for row in self._runs.values() + if row.owner_account_id == owner_account_id and row.deleted_at is None + ] owned.sort(key=lambda row: (row.queued_at, row.id), reverse=True) return tuple(owned[offset : offset + limit]) @@ -666,10 +679,50 @@ def transition(self, run_id: UUID, target: RunStatus, **values: Any) -> RunRow: raise InvalidStatusTransition( f"backtest run {run_id} is {current.status.value}; it cannot move to {target.value}" ) + if current.deletion_requested_at is not None and target in { + RunStatus.COMPLETED, + RunStatus.FAILED, + RunStatus.CANCELLED, + RunStatus.UNAVAILABLE, + }: + values.setdefault("deleted_at", values.get("completed_at") or values.get("cancelled_at")) updated = replace(current, status=target, **values) self._runs[run_id] = updated return updated + def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: + with self._lock: + current = self.get(run_id) + if current.deleted_at is not None: + return current + values: dict[str, Any] = { + "deletion_requested_at": current.deletion_requested_at or requested_at, + } + if current.status in { + RunStatus.COMPLETED, + RunStatus.FAILED, + RunStatus.CANCELLED, + RunStatus.UNAVAILABLE, + }: + values["deleted_at"] = requested_at + elif current.status is RunStatus.QUEUED: + values.update( + status=RunStatus.CANCELLED, + cancellation_requested_at=current.cancellation_requested_at or requested_at, + cancellation_reason_code=current.cancellation_reason_code or "USER_DELETED", + cancelled_at=requested_at, + completed_at=requested_at, + deleted_at=requested_at, + ) + else: + values.update( + cancellation_requested_at=current.cancellation_requested_at or requested_at, + cancellation_reason_code=current.cancellation_reason_code or "USER_DELETED", + ) + updated = replace(current, **values) + self._runs[run_id] = updated + return updated + def request_cancellation( self, run_id: UUID, *, reason_code: str, requested_at: datetime ) -> RunRow: @@ -926,6 +979,8 @@ def _build_run( def get(self, run_id: UUID, *, owner_account_id: UUID) -> BacktestRun: run = self._load(self.gateway.get(run_id)) self._require_owner(run, owner_account_id) + if run.run.deleted_at is not None: + raise BacktestRunNotFound(f"backtest run not found: {run_id}") return run def list_runs(self, owner_account_id: UUID, *, limit: int = 50, offset: int = 0) -> tuple[BacktestRun, ...]: @@ -987,6 +1042,27 @@ def request_cancellation( ) return self._load(row) + def request_deletion( + self, + run_id: UUID, + *, + owner_account_id: UUID, + requested_at: datetime | None = None, + ) -> BacktestRun: + """Hide a run from the owner while preserving durable execution evidence. + + Queued work is cancelled and deleted atomically. Running work receives a + cooperative cancellation request and becomes hidden only when it reaches a + terminal status. Repeating the request is idempotent. + """ + current = self._load(self.gateway.get(run_id)) + self._require_owner(current, owner_account_id) + row = self.gateway.request_deletion( + run_id, + requested_at=requested_at or datetime.now(timezone.utc), + ) + return self._load(row) + def _require_owner(self, run: BacktestRun, owner_account_id: UUID) -> None: if run.owner_account_id != owner_account_id: raise NotRunOwner(f"backtest run {run.backtest_run_id} belongs to another account") diff --git a/src/backtest_engine/orchestrator.py b/src/backtest_engine/orchestrator.py index 7c2706f..baa1e33 100644 --- a/src/backtest_engine/orchestrator.py +++ b/src/backtest_engine/orchestrator.py @@ -442,9 +442,11 @@ def __post_init__(self) -> None: if (self.evaluation_from is None) != (self.evaluation_through is None): raise OrchestratorError("evaluation interval must include both boundaries") if self.evaluation_from is not None: - if self.evaluation_from.tzinfo is None or self.evaluation_through.tzinfo is None: + evaluation_through = self.evaluation_through + assert evaluation_through is not None + if self.evaluation_from.tzinfo is None or evaluation_through.tzinfo is None: raise OrchestratorError("evaluation interval must be timezone-aware") - if self.evaluation_from >= self.evaluation_through: + if self.evaluation_from >= evaluation_through: raise OrchestratorError("evaluation interval must not be empty") # One source of truth for the bar period: the element catalog's # resolution table. A separately configured interval could disagree diff --git a/src/backtest_engine/persistence/protocols.py b/src/backtest_engine/persistence/protocols.py index d8eace3..f958742 100644 --- a/src/backtest_engine/persistence/protocols.py +++ b/src/backtest_engine/persistence/protocols.py @@ -99,6 +99,8 @@ def mark_failed(self, run_id: UUID, completed_at: datetime, failure_code: str) - def mark_unavailable(self, run_id: UUID, completed_at: datetime, failure_code: str) -> RunRow: ... + def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: ... + class AttemptStore(Protocol): """Durable replacement for the in-process attempt lock in `attempt_coordinator`.""" diff --git a/src/backtest_engine/persistence/repositories.py b/src/backtest_engine/persistence/repositories.py index 2ee31f1..c8f7764 100644 --- a/src/backtest_engine/persistence/repositories.py +++ b/src/backtest_engine/persistence/repositories.py @@ -194,7 +194,11 @@ def get_owned(self, owner_account_id: UUID, run_id: UUID) -> RunRow: """Owner-scoped read. A foreign run is indistinguishable from a missing one.""" found = self._fetch_one( - select(runs).where(runs.c.id == run_id, runs.c.owner_account_id == owner_account_id), + select(runs).where( + runs.c.id == run_id, + runs.c.owner_account_id == owner_account_id, + runs.c.deleted_at.is_(None), + ), RunRow, ) if found is None: @@ -222,7 +226,10 @@ def list_by_owner( raise ValueError("limit must be positive") if offset < 0: raise ValueError("offset must not be negative") - statement = select(runs).where(runs.c.owner_account_id == owner_account_id) + statement = select(runs).where( + runs.c.owner_account_id == owner_account_id, + runs.c.deleted_at.is_(None), + ) if bot_id is not None: statement = statement.where(runs.c.bot_id == bot_id) return self._fetch_all( @@ -336,7 +343,54 @@ def request_cancellation(self, run_id: UUID, *, reason_code: str) -> RunRow: ) return _hydrate(RunRow, updated) + def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: + """Cancel active work and retain the row as immutable owner evidence.""" + current = self._connection.execute( + select(runs).where(runs.c.id == run_id).with_for_update() + ).mappings().first() + if current is None: + raise RowNotFound(f"backtest run not found: {run_id}") + hydrated = _hydrate(RunRow, current) + if hydrated.deleted_at is not None: + return hydrated + values: dict[str, Any] = { + "deletion_requested_at": hydrated.deletion_requested_at or requested_at, + } + if hydrated.status in { + RunStatus.COMPLETED, + RunStatus.FAILED, + RunStatus.CANCELLED, + RunStatus.UNAVAILABLE, + }: + values["deleted_at"] = requested_at + elif hydrated.status is RunStatus.QUEUED: + values.update( + status=RunStatus.CANCELLED.value, + cancellation_requested_at=hydrated.cancellation_requested_at or requested_at, + cancellation_reason_code=hydrated.cancellation_reason_code or "USER_DELETED", + cancelled_at=requested_at, + completed_at=requested_at, + deleted_at=requested_at, + ) + else: + values.update( + cancellation_requested_at=hydrated.cancellation_requested_at or requested_at, + cancellation_reason_code=hydrated.cancellation_reason_code or "USER_DELETED", + ) + updated = self._connection.execute( + update(runs).where(runs.c.id == run_id).values(**values).returning(*runs.c) + ).mappings().one() + return _hydrate(RunRow, updated) + def _transition(self, run_id: UUID, target: RunStatus, **values: Any) -> RunRow: + current_before = self.get(run_id) + if current_before.deletion_requested_at is not None and target in { + RunStatus.COMPLETED, + RunStatus.FAILED, + RunStatus.CANCELLED, + RunStatus.UNAVAILABLE, + }: + values.setdefault("deleted_at", values.get("completed_at") or values.get("cancelled_at")) sources = sorted(source.value for source, allowed in RUN_STATUS_TRANSITIONS.items() if target in allowed) statement = ( update(runs) diff --git a/src/backtest_engine/persistence/rows.py b/src/backtest_engine/persistence/rows.py index 43c4aff..5985b3a 100644 --- a/src/backtest_engine/persistence/rows.py +++ b/src/backtest_engine/persistence/rows.py @@ -193,6 +193,8 @@ class RunRow: cancellation_requested_at: datetime | None = None cancellation_reason_code: str | None = None cancelled_at: datetime | None = None + deletion_requested_at: datetime | None = None + deleted_at: datetime | None = None def __post_init__(self) -> None: validate_money(self.initial_cash_amount, "initial_cash_amount") diff --git a/src/backtest_engine/persistence/tables.py b/src/backtest_engine/persistence/tables.py index f947ca9..e08f7a6 100644 --- a/src/backtest_engine/persistence/tables.py +++ b/src/backtest_engine/persistence/tables.py @@ -167,6 +167,8 @@ def _object_status() -> ENUM: Column("cancellation_requested_at", TIMESTAMP(timezone=True)), Column("cancellation_reason_code", VARCHAR(80)), Column("cancelled_at", TIMESTAMP(timezone=True)), + Column("deletion_requested_at", TIMESTAMP(timezone=True)), + Column("deleted_at", TIMESTAMP(timezone=True)), # Appended by the globally ordered forward contribution # `V20260805170000__backtest_run_outcome_detail.sql`. Each belongs to exactly one # terminal status and is NULL in every other, so absence stays distinguishable @@ -187,6 +189,7 @@ def _object_status() -> ENUM: Index("ix_runs_bot_id_queued_at", "bot_id", "queued_at"), Index("ix_runs_status_queued_at", "status", "queued_at"), Index("ix_runs_owner_account_id_queued_at", "owner_account_id", "queued_at"), + Index("ix_runs_owner_deleted_queued_at", "owner_account_id", "deleted_at", "queued_at"), schema=BACKTEST_SCHEMA, ) diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 53a6843..45d2577 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -1982,13 +1982,17 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: min(window[1] for window in scoped_windows), ) if plan.reference_series[1] == "$DATASET": - primary = [item for item in resolved_manifests if item[0].manifest_id == envelope.dataset_manifest_id] - if len(primary) != 1: + primary_pins = [ + item + for item in resolved_manifests + if item[0].manifest_id == envelope.dataset_manifest_id + ] + if len(primary_pins) != 1: raise JobNotSatisfiable( "the representative dataset is not pinned exactly once", reason_code="REQUIRED_INPUT_UNAVAILABLE", ) - manifest = primary[0][1] + manifest = primary_pins[0][1] dataset_resolution = str(manifest.get("resolution", "")) if dataset_resolution not in {"30m", "1h", "4h", "1d"}: raise JobNotSatisfiable( @@ -2013,12 +2017,12 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: f"the plan reference resolution {reference_resolution} has no pinned dataset cover", reason_code="REQUIRED_INPUT_UNAVAILABLE", ) - primary = [ + primary_manifests = [ resolved for pin, resolved in resolved_manifests if pin.manifest_id == envelope.dataset_manifest_id and str(resolved.get("resolution", "")) == reference_resolution ] - manifest = primary[0] if len(primary) == 1 else sorted( + manifest = primary_manifests[0] if len(primary_manifests) == 1 else sorted( reference_manifests, key=lambda item: dataset_coverage(item) )[0] if envelope.evaluation_start is not None or envelope.evaluation_end is not None: diff --git a/tests/persistence/test_roundtrip.py b/tests/persistence/test_roundtrip.py index 97452ab..165436b 100644 --- a/tests/persistence/test_roundtrip.py +++ b/tests/persistence/test_roundtrip.py @@ -241,6 +241,57 @@ def test_owner_scoped_read_hides_foreign_runs(persistence: BacktestPersistence) assert uow.runs.list_by_owner(ACCOUNT_ID) == (run,) +def test_owner_soft_delete_cancels_queued_run_and_preserves_internal_evidence( + persistence: BacktestPersistence, +) -> None: + run = make_run(idempotency_key="ROUNDTRIP:soft-delete") + requested_at = datetime(2026, 8, 27, 9, 0, tzinfo=UTC) + + with persistence.unit_of_work() as uow: + uow.runs.accept(run) + deleted = uow.runs.request_deletion(run.id, requested_at=requested_at) + + assert deleted.status is RunStatus.CANCELLED + assert deleted.cancellation_reason_code == "USER_DELETED" + assert deleted.deletion_requested_at == requested_at + assert deleted.deleted_at == requested_at + + with persistence.unit_of_work() as uow: + assert uow.runs.get(run.id).deleted_at == requested_at + with pytest.raises(RowNotFound): + uow.runs.get_owned(ACCOUNT_ID, run.id) + assert uow.runs.list_by_owner(ACCOUNT_ID) == () + + +def test_owner_soft_delete_waits_for_running_worker_then_hides_terminal_evidence( + persistence: BacktestPersistence, +) -> None: + run = make_run(idempotency_key="ROUNDTRIP:running-soft-delete") + started_at = datetime(2026, 8, 27, 9, 0, tzinfo=UTC) + requested_at = datetime(2026, 8, 27, 9, 1, tzinfo=UTC) + cancelled_at = datetime(2026, 8, 27, 9, 2, tzinfo=UTC) + + with persistence.unit_of_work() as uow: + uow.runs.accept(run) + uow.runs.mark_running(run.id, started_at) + pending = uow.runs.request_deletion(run.id, requested_at=requested_at) + + assert pending.status is RunStatus.RUNNING + assert pending.cancellation_requested_at == requested_at + assert pending.deletion_requested_at == requested_at + assert pending.deleted_at is None + + with persistence.unit_of_work() as uow: + cancelled = uow.runs.mark_cancelled(run.id, cancelled_at, "USER_DELETED") + + assert cancelled.status is RunStatus.CANCELLED + assert cancelled.deleted_at == cancelled_at + with persistence.unit_of_work() as uow: + with pytest.raises(RowNotFound): + uow.runs.get_owned(ACCOUNT_ID, run.id) + assert uow.runs.get(run.id).deleted_at == cancelled_at + + def test_lifecycle_transitions_use_the_canonical_completed_label( persistence: BacktestPersistence, ) -> None: diff --git a/tests/test_backtest_api.py b/tests/test_backtest_api.py index f54082c..2e5789b 100644 --- a/tests/test_backtest_api.py +++ b/tests/test_backtest_api.py @@ -520,6 +520,45 @@ def test_running_cancellation_is_cooperative_and_owner_scoped( assert run["cancelledAt"] is None +def test_owner_delete_is_idempotent_and_deleted_runs_disappear_from_customer_queries( + harness: Harness, official_request: dict[str, Any] +) -> None: + _accept(harness, official_request) + + first = harness.client.delete( + f"/api/v1/backtests/{EXPECTED_RUN_ID}", headers=harness.owner() + ) + second = harness.client.delete( + f"/api/v1/backtests/{EXPECTED_RUN_ID}", headers=harness.owner() + ) + + assert first.status_code == 202 + assert second.status_code == 202 + assert first.json()["run"]["status"] == "CANCELLED" + assert first.json()["deletionRequested"] is True + assert first.json()["deleted"] is True + assert harness.client.get("/api/v1/backtests", headers=harness.owner()).json()["items"] == [] + assert harness.client.get( + f"/api/v1/backtests/{EXPECTED_RUN_ID}", headers=harness.owner() + ).status_code == 404 + assert harness.gateway.get(UUID(EXPECTED_RUN_ID)).deleted_at is not None + + +def test_foreign_owner_cannot_delete_a_backtest( + harness: Harness, official_request: dict[str, Any] +) -> None: + _accept(harness, official_request) + + response = harness.client.delete( + f"/api/v1/backtests/{EXPECTED_RUN_ID}", headers=harness.other() + ) + + assert response.status_code == 403 + assert harness.client.get( + f"/api/v1/backtests/{EXPECTED_RUN_ID}", headers=harness.owner() + ).status_code == 200 + + def test_result_ingestion_requires_its_own_scope( harness: Harness, official_request: dict[str, Any] ) -> None: diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index 92162fc..bf17482 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -311,6 +311,53 @@ def test_an_unknown_run_is_not_found(service: BacktestLifecycleService) -> None: service.get(uuid4(), owner_account_id=OWNER_ID) +def test_deleting_a_queued_run_cancels_hides_and_preserves_its_evidence( + service: BacktestLifecycleService, official_request: dict[str, Any] +) -> None: + service.accept(official_request) + + deleted = service.request_deletion( + EXPECTED_RUN_ID, + owner_account_id=OWNER_ID, + requested_at=datetime(2026, 7, 31, 12, 4, tzinfo=timezone.utc), + ) + + assert deleted.status is RunStatus.CANCELLED + assert deleted.run.deletion_requested_at is not None + assert deleted.run.deleted_at is not None + assert service.list_runs(OWNER_ID) == () + with pytest.raises(BacktestRunNotFound): + service.get(EXPECTED_RUN_ID, owner_account_id=OWNER_ID) + assert service.gateway.get(EXPECTED_RUN_ID).id == EXPECTED_RUN_ID + + +def test_deleting_a_running_run_waits_for_cooperative_cancellation_before_hiding_evidence( + service: BacktestLifecycleService, official_request: dict[str, Any] +) -> None: + service.accept(official_request) + service.ingest_result(_event(service, "RUNNING", startedAt="2026-07-31T12:05:00Z", attempt=1)) + + pending = service.request_deletion( + EXPECTED_RUN_ID, + owner_account_id=OWNER_ID, + requested_at=datetime(2026, 7, 31, 12, 6, tzinfo=timezone.utc), + ) + assert pending.status is RunStatus.RUNNING + assert pending.run.deletion_requested_at is not None + assert pending.run.deleted_at is None + + terminal = service.ingest_result(_event( + service, + "CANCELLED", + cancelledAt="2026-07-31T12:07:00Z", + reasonCode="USER_DELETED", + attempt=1, + )).run + assert terminal.status is RunStatus.CANCELLED + assert terminal.run.deleted_at == datetime(2026, 7, 31, 12, 7, tzinfo=timezone.utc) + assert service.list_runs(OWNER_ID) == () + + # =========================================================================== # Result ingestion # =========================================================================== From fd016a64b0bbfd43b350bbfadbe0f88f98e99385 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Thu, 27 Aug 2026 13:13:27 +0900 Subject: [PATCH 4/9] fix(backtest): finalize concurrent owner deletion --- src/backtest_engine/lifecycle.py | 2 +- src/backtest_engine/persistence/protocols.py | 2 +- .../persistence/repositories.py | 22 ++++++++++---- src/backtest_engine/recovery.py | 5 +++- tests/persistence/test_roundtrip.py | 19 +++++------- tests/test_stale_recovery.py | 30 +++++++++++++++++++ 6 files changed, 61 insertions(+), 19 deletions(-) diff --git a/src/backtest_engine/lifecycle.py b/src/backtest_engine/lifecycle.py index 6d75195..5812f0d 100644 --- a/src/backtest_engine/lifecycle.py +++ b/src/backtest_engine/lifecycle.py @@ -548,7 +548,7 @@ def request_cancellation( def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: try: with self._write() as uow: - return uow.runs.request_deletion(run_id, requested_at=requested_at) + return uow.runs.request_deletion(run_id) except RowNotFound as exc: raise BacktestRunNotFound(str(exc)) from exc diff --git a/src/backtest_engine/persistence/protocols.py b/src/backtest_engine/persistence/protocols.py index f958742..1df3a89 100644 --- a/src/backtest_engine/persistence/protocols.py +++ b/src/backtest_engine/persistence/protocols.py @@ -99,7 +99,7 @@ def mark_failed(self, run_id: UUID, completed_at: datetime, failure_code: str) - def mark_unavailable(self, run_id: UUID, completed_at: datetime, failure_code: str) -> RunRow: ... - def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: ... + def request_deletion(self, run_id: UUID) -> RunRow: ... class AttemptStore(Protocol): diff --git a/src/backtest_engine/persistence/repositories.py b/src/backtest_engine/persistence/repositories.py index c8f7764..0080333 100644 --- a/src/backtest_engine/persistence/repositories.py +++ b/src/backtest_engine/persistence/repositories.py @@ -25,7 +25,7 @@ from typing import Any from uuid import UUID, uuid4 -from sqlalchemy import Connection, Row, Select, func, select, update +from sqlalchemy import Connection, Row, Select, case, func, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from .errors import ( @@ -343,7 +343,7 @@ def request_cancellation(self, run_id: UUID, *, reason_code: str) -> RunRow: ) return _hydrate(RunRow, updated) - def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: + def request_deletion(self, run_id: UUID) -> RunRow: """Cancel active work and retain the row as immutable owner evidence.""" current = self._connection.execute( select(runs).where(runs.c.id == run_id).with_for_update() @@ -353,6 +353,8 @@ def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: hydrated = _hydrate(RunRow, current) if hydrated.deleted_at is not None: return hydrated + requested_at = self._connection.scalar(select(func.clock_timestamp())) + assert isinstance(requested_at, datetime) values: dict[str, Any] = { "deletion_requested_at": hydrated.deletion_requested_at or requested_at, } @@ -383,14 +385,24 @@ def request_deletion(self, run_id: UUID, *, requested_at: datetime) -> RunRow: return _hydrate(RunRow, updated) def _transition(self, run_id: UUID, target: RunStatus, **values: Any) -> RunRow: - current_before = self.get(run_id) - if current_before.deletion_requested_at is not None and target in { + if target in { RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.CANCELLED, RunStatus.UNAVAILABLE, }: - values.setdefault("deleted_at", values.get("completed_at") or values.get("cancelled_at")) + terminal_at = values.get("completed_at") or values.get("cancelled_at") + if terminal_at is not None: + values.setdefault( + "deleted_at", + case( + ( + runs.c.deletion_requested_at.is_not(None), + func.greatest(terminal_at, runs.c.deletion_requested_at), + ), + else_=runs.c.deleted_at, + ), + ) sources = sorted(source.value for source, allowed in RUN_STATUS_TRANSITIONS.items() if target in allowed) statement = ( update(runs) diff --git a/src/backtest_engine/recovery.py b/src/backtest_engine/recovery.py index 971d639..d19ff46 100644 --- a/src/backtest_engine/recovery.py +++ b/src/backtest_engine/recovery.py @@ -101,7 +101,10 @@ def _recover_candidate( changed = connection.execute(text(""" UPDATE backtest.runs SET status='CANCELLED', completed_at=:now, cancelled_at=:now, - cancellation_reason_code=COALESCE(cancellation_reason_code, 'USER_CANCELLED') + cancellation_reason_code=COALESCE(cancellation_reason_code, 'USER_CANCELLED'), + deleted_at=CASE WHEN deletion_requested_at IS NOT NULL + THEN GREATEST(:now, deletion_requested_at) + ELSE deleted_at END WHERE id=:id AND status IN ('QUEUED', 'RUNNING') """), {"id": row["id"], "now": now}) counts["cancelled"] += changed.rowcount diff --git a/tests/persistence/test_roundtrip.py b/tests/persistence/test_roundtrip.py index 165436b..957801b 100644 --- a/tests/persistence/test_roundtrip.py +++ b/tests/persistence/test_roundtrip.py @@ -245,19 +245,17 @@ def test_owner_soft_delete_cancels_queued_run_and_preserves_internal_evidence( persistence: BacktestPersistence, ) -> None: run = make_run(idempotency_key="ROUNDTRIP:soft-delete") - requested_at = datetime(2026, 8, 27, 9, 0, tzinfo=UTC) - with persistence.unit_of_work() as uow: uow.runs.accept(run) - deleted = uow.runs.request_deletion(run.id, requested_at=requested_at) + deleted = uow.runs.request_deletion(run.id) assert deleted.status is RunStatus.CANCELLED assert deleted.cancellation_reason_code == "USER_DELETED" - assert deleted.deletion_requested_at == requested_at - assert deleted.deleted_at == requested_at + assert deleted.deletion_requested_at is not None + assert deleted.deleted_at == deleted.deletion_requested_at with persistence.unit_of_work() as uow: - assert uow.runs.get(run.id).deleted_at == requested_at + assert uow.runs.get(run.id).deleted_at == deleted.deleted_at with pytest.raises(RowNotFound): uow.runs.get_owned(ACCOUNT_ID, run.id) assert uow.runs.list_by_owner(ACCOUNT_ID) == () @@ -268,24 +266,23 @@ def test_owner_soft_delete_waits_for_running_worker_then_hides_terminal_evidence ) -> None: run = make_run(idempotency_key="ROUNDTRIP:running-soft-delete") started_at = datetime(2026, 8, 27, 9, 0, tzinfo=UTC) - requested_at = datetime(2026, 8, 27, 9, 1, tzinfo=UTC) cancelled_at = datetime(2026, 8, 27, 9, 2, tzinfo=UTC) with persistence.unit_of_work() as uow: uow.runs.accept(run) uow.runs.mark_running(run.id, started_at) - pending = uow.runs.request_deletion(run.id, requested_at=requested_at) + pending = uow.runs.request_deletion(run.id) assert pending.status is RunStatus.RUNNING - assert pending.cancellation_requested_at == requested_at - assert pending.deletion_requested_at == requested_at + assert pending.cancellation_requested_at is not None + assert pending.deletion_requested_at == pending.cancellation_requested_at assert pending.deleted_at is None with persistence.unit_of_work() as uow: cancelled = uow.runs.mark_cancelled(run.id, cancelled_at, "USER_DELETED") assert cancelled.status is RunStatus.CANCELLED - assert cancelled.deleted_at == cancelled_at + assert cancelled.deleted_at == max(cancelled_at, pending.deletion_requested_at) with persistence.unit_of_work() as uow: with pytest.raises(RowNotFound): uow.runs.get_owned(ACCOUNT_ID, run.id) diff --git a/tests/test_stale_recovery.py b/tests/test_stale_recovery.py index 3f6ba1f..fe8161e 100644 --- a/tests/test_stale_recovery.py +++ b/tests/test_stale_recovery.py @@ -157,6 +157,36 @@ def test_expired_running_cancellation_finishes_cancelled_not_failed( } +def test_expired_running_deletion_finishes_cancelled_and_hidden( + persistence: BacktestPersistence, admin_engine: Engine +) -> None: + run_id = _run(persistence) + store = PersistenceExecutionKeyStore(persistence) + store.claim( + worker_execution_key_for(str(run_id), "delete"), run_id=str(run_id), owner="worker", + now=datetime.now(UTC), lease_duration=timedelta(minutes=1), + ) + with persistence.unit_of_work() as uow: + pending = uow.runs.request_deletion(run_id) + assert pending.deletion_requested_at is not None + assert pending.deleted_at is None + _expire(admin_engine, run_id) + + report = StaleRunRecovery( + persistence, max_attempts=5, queued_timeout=timedelta(minutes=15) + ).recover_once() + + assert report.cancelled == 1 + with admin_engine.connect() as connection: + row = connection.execute(text(""" + SELECT status, deletion_requested_at, deleted_at + FROM backtest.runs WHERE id=:id + """), {"id": run_id}).mappings().one() + assert row["status"] == "CANCELLED" + assert row["deleted_at"] is not None + assert row["deleted_at"] >= row["deletion_requested_at"] + + def test_never_dispatched_queued_run_fails_after_timeout_and_recovery_is_idempotent( persistence: BacktestPersistence, admin_engine: Engine ) -> None: From f3b96bd96858584b2ea7a7d4a076b1450be926d6 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Thu, 27 Aug 2026 17:38:35 +0900 Subject: [PATCH 5/9] fix: execute mixed-resolution backtest flows --- src/backtest_engine/elements/orders.py | 2 +- src/backtest_engine/orchestrator.py | 20 +++- src/backtest_engine/wiring.py | 126 ++++++++++++++++++++----- tests/test_basic_runtime.py | 22 ++++- tests/test_elements_orders.py | 20 ++++ tests/test_orchestrator.py | 70 ++++++++++++++ tests/test_wiring.py | 100 ++++++++++++++++++++ 7 files changed, 328 insertions(+), 32 deletions(-) diff --git a/src/backtest_engine/elements/orders.py b/src/backtest_engine/elements/orders.py index 7886fca..c22f4cf 100644 --- a/src/backtest_engine/elements/orders.py +++ b/src/backtest_engine/elements/orders.py @@ -207,7 +207,7 @@ def __post_init__(self) -> None: allowed_modes = ( {"1회만", "주기마다", "대기 후 재진입"} if self.side == "BUY" - else {"1회만", "대기 후 재실행"} + else {"1회만", "주기마다", "대기 후 재실행"} ) if self.execution_mode not in allowed_modes: raise ElementEvaluationError( diff --git a/src/backtest_engine/orchestrator.py b/src/backtest_engine/orchestrator.py index baa1e33..799846a 100644 --- a/src/backtest_engine/orchestrator.py +++ b/src/backtest_engine/orchestrator.py @@ -488,6 +488,7 @@ def bar_events_from_batches( resolution: str, publication_lag: timedelta = timedelta(0), schedule: OfficialSessionSchedule | None = None, + allow_empty: bool = False, ) -> tuple[MarketDataEvent, ...]: """Turn verified Parquet rows into the clock's event stream. @@ -560,7 +561,7 @@ def bar_events_from_batches( }, ) ) - if not events: + if not events and not allow_empty: raise OrchestratorError("the pinned dataset contains no bars") return tuple(events) @@ -773,6 +774,12 @@ def run( resolution=resolution, publication_lag=self._publication_lag, schedule=schedule, + # Universe manifests are segmented by time. An instrument + # may legitimately have no rows in an early segment (for + # example before listing) while later pinned segments do. + # Availability is assessed over the combined verified + # stream below, not one storage segment at a time. + allow_empty=True, ) combined_events.extend( replace(event, event_id=f"{event.event_id}:{resolution}") @@ -915,11 +922,20 @@ def _execute( ) try: evaluation = replay.evaluate_at(instant, visible_events, runtime_values) - except Exception: + except Exception as exc: + detail = f"{type(exc).__name__}: {exc}" + _LOG.error( + "backtest run %s plan replay failed at %s: %s", + job.run_id, + instant.isoformat(), + detail, + exc_info=exc, + ) return self._abort( job, coordinator, lease, "PLAN_REPLAY_FAILED", retryable=False, status=ReplayStatus.FAILED, availability=assessment.status, steps=tuple(steps), + detail=detail, ) compact_evaluation = getattr(replay, "compact_evaluation", None) evaluations.append( diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 45d2577..8ea5087 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -98,7 +98,7 @@ ReplayLedgerDetail, ) from .elements import PinnedFeatureSeries -from .event_clock import MarketDataEvent, MarketEventClock +from .event_clock import MarketDataEvent, MarketEventClock, OfficialSessionSchedule from .execution_model import ( BacktestExecutionModel, ExecutionBar, @@ -1613,6 +1613,78 @@ def segmented_dataset_coverage( return start, end +def _dataset_cover_contains_evaluation( + schedule: OfficialSessionSchedule, + coverage_start: datetime, + coverage_end: datetime, + evaluation_from: datetime, + evaluation_through: datetime, +) -> bool: + """Accept wall-clock boundary gaps only when no regular session is omitted. + + Dataset object boundaries describe the first and last delivered bars, while + an explicit evaluation request uses whole local trading dates. Those two + clocks legitimately differ overnight, on weekends, and on holidays. The + cover is sufficient when every official session that intersects the + evaluation interval is fully contained by it; an actually missing trading + session still fails closed. + """ + if coverage_start >= coverage_end or evaluation_from >= evaluation_through: + return False + for session in schedule.sessions: + if session.closes_at <= evaluation_from or session.opens_at >= evaluation_through: + continue + required_start = max(session.opens_at, evaluation_from) + required_end = min(session.closes_at, evaluation_through) + if required_start < coverage_start or required_end > coverage_end: + return False + return True + + +def _resolve_position_only_flow_clocks( + plan: BasicCompiledPlan, *, fallback_resolution: str | None = None +) -> BasicCompiledPlan: + """Bind each position-only flow to the unique market clock for its instruments.""" + has_concrete_clock = any( + flow.reference_series[1] != "$DATASET" for flow in plan.flows + ) + if not has_concrete_clock: + if fallback_resolution is None: + raise JobNotSatisfiable( + "a position-only plan has no market clock", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + return replace( + plan, + flows=tuple( + replace( + flow, + reference_series=(flow.reference_series[0], fallback_resolution), + ) + for flow in plan.flows + ), + ) + + resolved_flows = [] + for flow in plan.flows: + if flow.reference_series[1] != "$DATASET": + resolved_flows.append(flow) + continue + inherited = { + candidate.reference_series + for candidate in plan.flows + if candidate.reference_series[1] != "$DATASET" + and set(candidate.instrument_ids).intersection(flow.instrument_ids) + } + if len(inherited) != 1: + raise JobNotSatisfiable( + f"position-only flow {flow.flow_id} has no unambiguous market clock", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + resolved_flows.append(replace(flow, reference_series=next(iter(inherited)))) + return replace(plan, flows=tuple(resolved_flows)) + + def evaluation_window(manifest: Mapping[str, Any], plan: BasicCompiledPlan) -> tuple[datetime, datetime]: """Where warm-up ends and evaluation begins, for this plan on this dataset. @@ -1981,6 +2053,7 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: max(window[0] for window in scoped_windows), min(window[1] for window in scoped_windows), ) + position_only_fallback: str | None = None if plan.reference_series[1] == "$DATASET": primary_pins = [ item @@ -2002,13 +2075,8 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: plan = replace( plan, reference_series=(plan.reference_series[0], dataset_resolution), - flows=tuple( - replace(flow, reference_series=(flow.reference_series[0], dataset_resolution)) - if flow.reference_series[1] == "$DATASET" - else flow - for flow in plan.flows - ), ) + position_only_fallback = dataset_resolution else: reference_resolution = plan.reference_series[1] reference_manifests = manifests_by_resolution.get(reference_resolution, []) @@ -2025,6 +2093,10 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: manifest = primary_manifests[0] if len(primary_manifests) == 1 else sorted( reference_manifests, key=lambda item: dataset_coverage(item) )[0] + plan = _resolve_position_only_flow_clocks( + plan, + fallback_resolution=position_only_fallback, + ) if envelope.evaluation_start is not None or envelope.evaluation_end is not None: if envelope.evaluation_start is None or envelope.evaluation_end is None: raise JobNotSatisfiable( @@ -2044,32 +2116,34 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: reason_code="REQUIRED_INPUT_UNAVAILABLE", ) else: - resolved_flows = [] - for flow in plan.flows: - if flow.reference_series[1] != "$DATASET": - resolved_flows.append(flow) - continue - inherited = { - candidate.reference_series - for candidate in plan.flows - if candidate.reference_series[1] != "$DATASET" - and set(candidate.instrument_ids).intersection(flow.instrument_ids) - } - if len(inherited) != 1: - raise JobNotSatisfiable( - f"position-only flow {flow.flow_id} has no unambiguous market clock", - reason_code="REQUIRED_INPUT_UNAVAILABLE", - ) - resolved_flows.append(replace(flow, reference_series=next(iter(inherited)))) - plan = replace(plan, flows=tuple(resolved_flows)) reference_start, reference_end = coverage_by_resolution[plan.reference_series[1]] warmup = max( (feature.warmup_span for feature in plan.required_features), default=timedelta(0), ) evaluation_from, evaluation_through = reference_start + warmup, reference_end + policy_zone = ZoneInfo(policy.timezone) + schedule_first = evaluation_from.astimezone(policy_zone).date() + schedule_last = (evaluation_through - timedelta(microseconds=1)).astimezone( + policy_zone + ).date() + try: + evaluation_schedule = self._calendar.session_schedule( + schedule_first, schedule_last + ) + except CalendarCoverageError as exc: + raise JobNotSatisfiable( + f"the official session calendar does not cover the evaluation interval: {exc}", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) from exc for resolution, (coverage_start, coverage_end) in coverage_by_resolution.items(): - if coverage_start > evaluation_from or coverage_end < evaluation_through: + if not _dataset_cover_contains_evaluation( + evaluation_schedule, + coverage_start, + coverage_end, + evaluation_from, + evaluation_through, + ): raise JobNotSatisfiable( f"pinned {resolution} dataset cover {coverage_start.isoformat()}.." f"{coverage_end.isoformat()} does not contain evaluation interval " diff --git a/tests/test_basic_runtime.py b/tests/test_basic_runtime.py index 3c34063..138d3f6 100644 --- a/tests/test_basic_runtime.py +++ b/tests/test_basic_runtime.py @@ -64,6 +64,7 @@ FIRST = "00000000-0000-4000-8000-000000000301" SECOND = "00000000-0000-4000-8000-000000000302" +THIRD = "00000000-0000-4000-8000-000000000303" SESSION_DATE = date(2025, 11, 28) OPEN = datetime.fromisoformat("2025-11-28T14:30:00+00:00") @@ -1441,10 +1442,13 @@ def test_loads_raw_market_containers_with_independent_resolutions() -> None: document["elementCatalogVersion"] = "basic-elements:2026-08-25" document["requiredFeatures"] = [] flows = document["executionSnapshot"]["partitions"][0]["flows"] + daily_flow = copy.deepcopy(flows[-1]) + daily_flow["key"] = "daily-flow" + flows.append(daily_flow) for flow, instrument_id, resolution in zip( flows, - (FIRST, SECOND), - ("1h", "4h"), + (FIRST, SECOND, THIRD), + ("30m", "4h", "1d"), strict=True, ): flow["officialInstrumentIds"] = [instrument_id] @@ -1476,8 +1480,20 @@ def test_loads_raw_market_containers_with_independent_resolutions() -> None: plan = _runtime().load(document) assert [flow.reference_series for flow in plan.flows] == [ - ("ADJUSTED_BAR", "1h"), + ("ADJUSTED_BAR", "30m"), ("ADJUSTED_BAR", "4h"), + ("ADJUSTED_BAR", "1d"), + ] + + requirements = derive_data_requirements( + plan, + evaluation_from=_utc("2025-11-28T14:45:00Z"), + evaluation_through=_utc("2025-11-28T20:00:00Z"), + ) + assert [(item.instrument_id, item.resolution) for item in requirements] == [ + (FIRST, "30m"), + (SECOND, "4h"), + (THIRD, "1d"), ] diff --git a/tests/test_elements_orders.py b/tests/test_elements_orders.py index 7e10664..231fdee 100644 --- a/tests/test_elements_orders.py +++ b/tests/test_elements_orders.py @@ -135,6 +135,26 @@ def test_the_side_comes_from_the_step_and_a_sell_carries_no_allocation() -> None assert candidate.allocation is None +def test_a_sell_can_execute_each_period_as_published_by_the_shared_catalog() -> None: + candidate = _emit( + step=_step( + allocation="EQUAL", + orderType="MARKET", + side="SELL", + orderPercent="100", + maxPositionPercent="25", + executionMode="주기마다", + waitMode="조건 재충족", + waitInterval="1", + maxExecutions="100", + ), + allocation=None, + ) + + assert candidate.side == "SELL" + assert candidate.execution_mode == "주기마다" + + def test_a_non_terminal_step_can_never_emit() -> None: load = PlanStep( sequence=1, diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index 28fb08e..c5b8430 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -691,6 +691,75 @@ def iter_batches( assert {series.resolution for series in visible[MSFT].series} == {"30m"} +def test_orchestrator_allows_an_empty_segment_when_a_later_segment_has_required_bars() -> None: + rows = pa.Table.from_pylist( + [ + _bar_row(_utc(14, 30), date(2024, 1, 2)), + _bar_row(_utc(14, 45), date(2024, 1, 2)), + ], + schema=_SCHEMA, + ) + manifests = ( + {"resolution": "15m", "segment": "before-listing"}, + {"resolution": "15m", "segment": "available"}, + ) + + class SegmentedReader: + def iter_batches( + self, + manifest: Mapping[str, Any], + _policy: Any, + *, + instrument_ids: frozenset[str] | None = None, + ) -> Any: + assert instrument_ids == frozenset({AAPL}) + return () if manifest["segment"] == "before-listing" else rows.to_batches() + + runtime = StubRuntime(buy_at=None) + harness = Harness(Path(), runtime, RecordingEngine(), RecordingPublisher()) + orchestrator = BacktestOrchestrator( + reader=SegmentedReader(), + calendar=XNYS_CALENDAR, + replay_factory=harness.factory, + engine=harness.engine, + publisher=harness.publisher, + wall_clock=WallClock(), + ) + requirement = DataRequirement( + requirement_id="aapl-15m", + instrument_id=AAPL, + data_kind=DATA_KIND, + resolution="15m", + warmup_from=_utc(14, 30), + evaluation_from=_utc(14, 30), + evaluation_through=_utc(15, 0), + ) + coordinator = AttemptCoordinator(RUN_ID, _policy(), WALL_T0) + + outcome = orchestrator.run( + BacktestJob( + run_id=RUN_ID, + idempotency_key="OFFICIAL_BACKTEST:segmented", + worker_execution_key=f"BACKTEST_RUN:{RUN_ID}:segmented", + manifest=manifests[0], + execution_policy=D17_EXECUTION_POLICY_FIXTURE, + requirements=(requirement,), + data_kind=DATA_KIND, + resolution="15m", + initial_cash=Decimal("10000"), + manifests=manifests, + evaluation_from=_utc(14, 30), + evaluation_through=_utc(15, 0), + ), + coordinator=coordinator, + lease=coordinator.acquire("segmented-worker", WALL_T0), + monitor=FixedMonitor(), + ) + + assert outcome.status is ReplayStatus.COMPLETED + assert len(outcome.steps) == 1 + + # -------------------------------------------------------------------------- # The replay loop is genuinely driven by the event clock. # -------------------------------------------------------------------------- @@ -870,6 +939,7 @@ def test_plan_replay_failure_fails_the_run_permanently_and_publishes_nothing( assert outcome.status is ReplayStatus.FAILED assert outcome.reason_code == "PLAN_REPLAY_FAILED" + assert outcome.failure_detail == "ZeroDivisionError: condition evaluator blew up" assert harness.publisher.requests == [] assert coordinator.state is RunState.FAILED assert coordinator.attempts[0].state is AttemptState.PERMANENT_FAILED diff --git a/tests/test_wiring.py b/tests/test_wiring.py index 9516f6a..8905fe7 100644 --- a/tests/test_wiring.py +++ b/tests/test_wiring.py @@ -17,6 +17,7 @@ import copy import logging import uuid +from dataclasses import replace from datetime import UTC, date, datetime, timedelta from decimal import ROUND_HALF_UP, Decimal, localcontext from fractions import Fraction @@ -67,7 +68,9 @@ OrchestratorJobHandler, WiringError, _CancellationAwareMonitor, + _dataset_cover_contains_evaluation, _metric_percent, + _resolve_position_only_flow_clocks, dataset_coverage, evaluation_window, segmented_dataset_coverage, @@ -680,6 +683,82 @@ def test_adjacent_manifests_of_the_same_resolution_form_one_cover() -> None: ) +def test_position_only_flow_inherits_its_own_instruments_clock_in_a_mixed_resolution_plan() -> None: + plan = _plan() + source = plan.flows[0] + other_instrument = "00000000-0000-4000-8000-000000000099" + thirty_minute = replace( + source, + flow_id="aapl-30m", + instrument_ids=(INSTRUMENT_ID,), + reference_series=("ADJUSTED_BAR", "30m"), + ) + four_hour = replace( + source, + flow_id="meta-4h", + instrument_ids=(other_instrument,), + reference_series=("ADJUSTED_BAR", "4h"), + ) + position_only = replace( + source, + flow_id="meta-position-exit", + instrument_ids=(other_instrument,), + condition_steps=(), + reference_series=("ADJUSTED_BAR", "$DATASET"), + ) + mixed = replace( + plan, + reference_series=thirty_minute.reference_series, + flows=(thirty_minute, four_hour, position_only), + ) + + resolved = _resolve_position_only_flow_clocks(mixed) + + assert [flow.reference_series for flow in resolved.flows] == [ + ("ADJUSTED_BAR", "30m"), + ("ADJUSTED_BAR", "4h"), + ("ADJUSTED_BAR", "4h"), + ] + + +def test_position_only_first_flow_uses_its_instruments_clock_not_representative_dataset() -> None: + plan = _plan() + source = plan.flows[0] + meta = "00000000-0000-4000-8000-000000000099" + position_only = replace( + source, + flow_id="meta-position-exit", + instrument_ids=(meta,), + condition_steps=(), + reference_series=("ADJUSTED_BAR", "$DATASET"), + ) + meta_four_hour = replace( + source, + flow_id="meta-4h-entry", + instrument_ids=(meta,), + reference_series=("ADJUSTED_BAR", "4h"), + ) + aapl_thirty_minute = replace( + source, + flow_id="aapl-30m-entry", + instrument_ids=(INSTRUMENT_ID,), + reference_series=("ADJUSTED_BAR", "30m"), + ) + mixed = replace( + plan, + reference_series=("ADJUSTED_BAR", "$DATASET"), + flows=(position_only, meta_four_hour, aapl_thirty_minute), + ) + + resolved = _resolve_position_only_flow_clocks(mixed, fallback_resolution="30m") + + assert [flow.reference_series for flow in resolved.flows] == [ + ("ADJUSTED_BAR", "4h"), + ("ADJUSTED_BAR", "4h"), + ("ADJUSTED_BAR", "30m"), + ] + + def test_segmented_manifest_cover_rejects_a_gap() -> None: first = dataset_manifest("1" * 64, row_count=1, coverage_end=FIRST_BAR_START + BAR) second = dataset_manifest("2" * 64, row_count=1, coverage_end=FIRST_BAR_START + BAR) @@ -694,6 +773,27 @@ def test_segmented_manifest_cover_rejects_a_gap() -> None: segmented_dataset_coverage((first, second)) +def test_dataset_cover_may_end_before_local_midnight_only_when_no_session_is_missing() -> None: + schedule = XNYS_CALENDAR.session_schedule(date(2026, 7, 29), date(2026, 7, 30)) + coverage_start = datetime(2016, 1, 1, tzinfo=UTC) + coverage_end = datetime(2026, 7, 30, tzinfo=UTC) + + assert _dataset_cover_contains_evaluation( + schedule, + coverage_start, + coverage_end, + datetime(2016, 1, 1, 5, tzinfo=UTC), + datetime(2026, 7, 30, 4, tzinfo=UTC), + ) + assert not _dataset_cover_contains_evaluation( + schedule, + coverage_start, + coverage_end, + datetime(2016, 1, 1, 5, tzinfo=UTC), + datetime(2026, 7, 31, 4, tzinfo=UTC), + ) + + def test_the_pinned_completion_instant_follows_every_replay_instant() -> None: """Guards the fixture: `completed_at` must not precede a result record.""" assert COMPLETED_AT > FIRST_BAR_START + BAR * len(CLOSES) From 1a18f2fc4cfffdbad40d661f9e37ff9f9e23b3b7 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Thu, 27 Aug 2026 17:49:34 +0900 Subject: [PATCH 6/9] fix: validate multi-manifest custom requests --- .../backtest_request_intake.py | 8 +++++++ tests/test_backtest_request_intake.py | 24 +++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/src/backtest_engine/backtest_request_intake.py b/src/backtest_engine/backtest_request_intake.py index 30e1dc1..5cc9fa5 100644 --- a/src/backtest_engine/backtest_request_intake.py +++ b/src/backtest_engine/backtest_request_intake.py @@ -594,6 +594,14 @@ def validate_backtest_request(document: Mapping[str, Any]) -> dict[str, Any]: "instrumentCatalogVersion", ) ] + if "datasets" in instance: + request_fields.append( + "".join( + f"{dataset['datasetManifestId']}:{dataset['purposeCode']}:" + f"{dataset['expectedDatasetHash']};" + for dataset in instance["datasets"] + ) + ) request_fields.append("".join( f"{feature['featureMaterializationId']}:{feature['lockedResultHash']};" for feature in sorted( diff --git a/tests/test_backtest_request_intake.py b/tests/test_backtest_request_intake.py index b762fdc..24433c2 100644 --- a/tests/test_backtest_request_intake.py +++ b/tests/test_backtest_request_intake.py @@ -113,6 +113,30 @@ def custom_request_with_two_market_datasets() -> dict[str, Any]: "expectedDatasetHash": "sha256:" + "9" * 64, }, ] + dataset_material = "".join( + f"{item['datasetManifestId']}:{item['purposeCode']}:{item['expectedDatasetHash']};" + for item in request["datasets"] + ) + request["requestHash"] = _sha( + "\n".join( + ( + request["requestingAccountId"], + request["botId"], + request["datasetManifestId"], + request["expectedDatasetHash"], + request["periodStart"], + request["periodEnd"], + request["expectedSnapshotHash"], + request["compiledPlanChecksum"], + request["instrumentCatalogVersion"], + dataset_material, + "", + request["initialCashAmount"], + request["assumptionsVersion"], + request["executionPolicyVersion"], + ) + ) + ) return request From bf71beb0da3d82739677d81b615eefa6a1bb97f3 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Fri, 28 Aug 2026 12:15:25 +0900 Subject: [PATCH 7/9] fix: validate all basic blocks and mixed fills --- src/backtest_engine/elements/catalog.py | 139 ++++++-- src/backtest_engine/feature_outputs.py | 40 ++- src/backtest_engine/result_snapshot.py | 95 +++-- src/backtest_engine/wiring.py | 25 +- tests/test_basic_element_conformance.py | 441 +++++++++++++++++++++++- tests/test_basic_runtime.py | 20 +- tests/test_feature_outputs.py | 43 ++- tests/test_result_snapshot.py | 47 +++ tests/test_wiring.py | 17 +- 9 files changed, 771 insertions(+), 96 deletions(-) diff --git a/src/backtest_engine/elements/catalog.py b/src/backtest_engine/elements/catalog.py index 1589f67..eb93ddb 100644 --- a/src/backtest_engine/elements/catalog.py +++ b/src/backtest_engine/elements/catalog.py @@ -43,7 +43,7 @@ import operator as operator_module from collections.abc import Callable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field, replace from decimal import ROUND_HALF_UP, Context, Decimal, InvalidOperation from types import MappingProxyType @@ -228,12 +228,31 @@ def _value(evaluation: ElementEvaluation, key: str, operation: str) -> str: def _decimal_value(evaluation: ElementEvaluation, key: str, operation: str) -> Decimal: - return Decimal(_value(evaluation, key, operation)) + raw = _value(evaluation, key, operation) + try: + value = Decimal(raw) + except InvalidOperation as exc: + raise ElementEvaluationError(f"{operation} runtime input {key} must be a decimal number, got {raw!r}") from exc + if not value.is_finite(): + raise ElementEvaluationError(f"{operation} runtime input {key} must be finite, got {raw!r}") + return value def _series_values(evaluation: ElementEvaluation, key: str, required: int, operation: str) -> list[Decimal]: raw = _value(evaluation, key, operation) - values = [Decimal(item) for item in raw.split(",") if item] + values: list[Decimal] = [] + for item in raw.split(","): + if not item: + continue + try: + value = Decimal(item) + except InvalidOperation as exc: + raise ElementEvaluationError( + f"{operation} runtime input {key} must contain decimal numbers, got {item!r}" + ) from exc + if not value.is_finite(): + raise ElementEvaluationError(f"{operation} runtime input {key} must contain finite numbers, got {item!r}") + values.append(value) if len(values) < required: raise ElementInputMissing( f"{operation} needs {required} values from {key}, got {len(values)}", @@ -352,10 +371,7 @@ def _evaluate_catalog_operation(step: PlanStep, evaluation: ElementEvaluation) - if ( resolution is not None and operation not in {"HOLDING_PERIOD", "SCHEDULE"} - and evaluation.inputs.values.get( - f"bar.closed.{resolution}", "false" - ).lower() - != "true" + and evaluation.inputs.values.get(f"bar.closed.{resolution}", "false").lower() != "true" ): return StepOutcome.failed( "WAITING_FOR_BAR_CLOSE", @@ -412,9 +428,7 @@ def _evaluate_catalog_operation(step: PlanStep, evaluation: ElementEvaluation) - closes = _series_values(evaluation, f"closes.{resolution}", bars + 1, operation) direction = step.argument("direction") count = 0 - for current, previous in zip( - reversed(closes[1:]), reversed(closes[:-1]), strict=True - ): + for current, previous in zip(reversed(closes[1:]), reversed(closes[:-1]), strict=True): if (direction == "UP" and current > previous) or (direction == "DOWN" and current < previous): count += 1 else: @@ -466,13 +480,8 @@ def _evaluate_catalog_operation(step: PlanStep, evaluation: ElementEvaluation) - int(step.argument("signalPeriod")), ) closes = _series_values(evaluation, f"closes.{resolution}", slow + signal + 2, operation) - macd = [ - a - b - for a, b in zip(_ema(closes, fast), _ema(closes, slow), strict=True) - ] - histogram = [ - a - b for a, b in zip(macd, _ema(macd, signal), strict=True) - ] + macd = [a - b for a, b in zip(_ema(closes, fast), _ema(closes, slow), strict=True)] + histogram = [a - b for a, b in zip(macd, _ema(macd, signal), strict=True)] previous, current = histogram[-2], histogram[-1] direction = step.argument("direction") passed = previous <= 0 < current if direction == "UP" else previous >= 0 > current @@ -569,6 +578,10 @@ class ElementSpec: produces_value: bool consumes_value: bool evaluator: StepEvaluator + minimum_arguments: Mapping[str, Decimal] = field(default_factory=lambda: MappingProxyType({})) + exclusive_minimum_arguments: Mapping[str, Decimal] = field(default_factory=lambda: MappingProxyType({})) + maximum_arguments: Mapping[str, Decimal] = field(default_factory=lambda: MappingProxyType({})) + integer_arguments: tuple[str, ...] = () @dataclass(frozen=True, slots=True) @@ -629,8 +642,31 @@ def validate_step(self, step: PlanStep) -> ElementSpec: PlanLoadFailure.UNSUPPORTED_ELEMENT_ARGUMENT, f"{step.operation} argument {name}={value!r} is not one of " + ", ".join(allowed), ) - for name in spec.decimal_arguments: - _parse_decimal(step, name) + decimal_values = {name: _parse_decimal(step, name) for name in spec.decimal_arguments} + for name, minimum in spec.minimum_arguments.items(): + if decimal_values[name] < minimum: + raise _reject( + PlanLoadFailure.UNSUPPORTED_ELEMENT_ARGUMENT, + f"{step.operation} argument {name} must be at least {minimum}, got {step.arguments[name]!r}", + ) + for name, minimum in spec.exclusive_minimum_arguments.items(): + if decimal_values[name] <= minimum: + raise _reject( + PlanLoadFailure.UNSUPPORTED_ELEMENT_ARGUMENT, + f"{step.operation} argument {name} must be greater than {minimum}, got {step.arguments[name]!r}", + ) + for name, maximum in spec.maximum_arguments.items(): + if decimal_values[name] > maximum: + raise _reject( + PlanLoadFailure.UNSUPPORTED_ELEMENT_ARGUMENT, + f"{step.operation} argument {name} must be at most {maximum}, got {step.arguments[name]!r}", + ) + for name in spec.integer_arguments: + if decimal_values[name] != decimal_values[name].to_integral_value(): + raise _reject( + PlanLoadFailure.UNSUPPORTED_ELEMENT_ARGUMENT, + f"{step.operation} argument {name} must be an integer, got {step.arguments[name]!r}", + ) for name in spec.feature_arguments: self.require_feature(step.arguments[name]) return spec @@ -957,6 +993,22 @@ def _production_spec( def _v2_specs() -> Mapping[str, ElementSpec]: specs = dict(_BASIC_ELEMENTS_2026_08_08.specs) + zero = Decimal(0) + hundred = Decimal(100) + specs["PRICE_CHANGE_PERCENT"] = replace( + specs["PRICE_CHANGE_PERCENT"], + minimum_arguments=MappingProxyType({"thresholdPercent": zero}), + ) + specs["RSI_CROSS"] = replace( + specs["RSI_CROSS"], + minimum_arguments=MappingProxyType({"threshold": zero}), + maximum_arguments=MappingProxyType({"threshold": hundred}), + ) + specs["POSITION_RETURN"] = replace( + specs["POSITION_RETURN"], + minimum_arguments=MappingProxyType({"thresholdPercent": zero}), + maximum_arguments=MappingProxyType({"thresholdPercent": hundred}), + ) specs["HOLDING_PERIOD"] = _production_spec( "HOLDING_PERIOD", ("unit", "amount", "resolution"), @@ -966,6 +1018,22 @@ def _v2_specs() -> Mapping[str, ElementSpec]: }, decimals=("amount",), ) + specs["HOLDING_PERIOD"] = replace( + specs["HOLDING_PERIOD"], + minimum_arguments=MappingProxyType({"amount": zero}), + integer_arguments=("amount",), + ) + for operation in ("PEAK_RETURN", "DRAWDOWN_FROM_PEAK"): + specs[operation] = replace( + specs[operation], + minimum_arguments=MappingProxyType({"thresholdPercent": zero}), + maximum_arguments=MappingProxyType({"thresholdPercent": hundred}), + ) + specs["SCHEDULE"] = replace( + specs["SCHEDULE"], + exclusive_minimum_arguments=MappingProxyType({"interval": zero}), + integer_arguments=("interval",), + ) specs["EMIT_ORDER_CANDIDATE"] = ElementSpec( operation="EMIT_ORDER_CANDIDATE", required_arguments=( @@ -1001,6 +1069,21 @@ def _v2_specs() -> Mapping[str, ElementSpec]: produces_value=False, consumes_value=False, evaluator=_evaluate_terminal, + exclusive_minimum_arguments=MappingProxyType( + { + "orderPercent": zero, + "maxPositionPercent": zero, + "waitInterval": zero, + "maxExecutions": zero, + } + ), + maximum_arguments=MappingProxyType( + { + "orderPercent": hundred, + "maxPositionPercent": hundred, + } + ), + integer_arguments=("waitInterval", "maxExecutions"), ) return MappingProxyType(specs) @@ -1009,8 +1092,22 @@ def _v2_specs() -> Mapping[str, ElementSpec]: version="basic-elements:2026-08-25", specs=_v2_specs(), feature_versions=_BASIC_ELEMENTS_2026_08_08.feature_versions, - canonical_feature_ids=_BASIC_ELEMENTS_2026_08_08.canonical_feature_ids, - canonical_feature_resolutions=_BASIC_ELEMENTS_2026_08_08.canonical_feature_resolutions, + canonical_feature_ids=MappingProxyType( + { + "ec37984b-6605-5560-8ea0-774c5b8e9626": "RSI_14", + "85f4f80f-be4e-d9dc-bd52-d4781ba5f30f": "RSI_14", + "65a5aaf5-f536-820f-119a-239b0aec0de7": "RSI_14", + "647a5fd6-98ed-0617-d4b2-844748d54fac": "RSI_14", + } + ), + canonical_feature_resolutions=MappingProxyType( + { + "ec37984b-6605-5560-8ea0-774c5b8e9626": "30m", + "85f4f80f-be4e-d9dc-bd52-d4781ba5f30f": "1h", + "65a5aaf5-f536-820f-119a-239b0aec0de7": "4h", + "647a5fd6-98ed-0617-d4b2-844748d54fac": "1d", + } + ), ) diff --git a/src/backtest_engine/feature_outputs.py b/src/backtest_engine/feature_outputs.py index 9172749..a17a043 100644 --- a/src/backtest_engine/feature_outputs.py +++ b/src/backtest_engine/feature_outputs.py @@ -77,6 +77,30 @@ "6d2647f8-5caf-55ee-8821-869dc693f68a", "FEATURE_RSI_14_1D_RSI_1_0_0", ), + "ec37984b-6605-5560-8ea0-774c5b8e9626": ( + "sha256:250df12e46d233e7b8ece86c64df7a3941f0d70436aebe522b1387f15fb346dc", + "30m", + "57794d8c-2254-53e4-966e-44f97edd9e6a", + "FEATURE_RSI_14_30M_RSI_1_0_0", + ), + "85f4f80f-be4e-d9dc-bd52-d4781ba5f30f": ( + "sha256:7e8c5600ff2bf07a043f797a50d6467f86fbdb56ee532c87929df97f246af2de", + "1h", + "28012549-4f45-56d3-8bb6-329e4c7a9d77", + "FEATURE_RSI_14_1H_RSI_1_0_0", + ), + "65a5aaf5-f536-820f-119a-239b0aec0de7": ( + "sha256:42e28b02a1552eb2aa42e0d89b1ea3dd909ee8d34c3bc290c4ce0234c6d705da", + "4h", + "e1d7d508-aaf1-5ae9-8098-c4af870f6fa4", + "FEATURE_RSI_14_4H_RSI_1_0_0", + ), + "647a5fd6-98ed-0617-d4b2-844748d54fac": ( + "sha256:64dbbcda7352d0add9a4a6a6ed94a780603880891684dc32cf39e0a3d1167422", + "1d", + "6d2647f8-5caf-55ee-8821-869dc693f68a", + "FEATURE_RSI_14_1D_RSI_1_0_0", + ), } FEATURE_SERIES_SCHEMA = pa.schema( [ @@ -149,18 +173,6 @@ def _expected_feature_feed_id(record: Mapping[str, Any]) -> str: ) if resolution != expected_resolution: raise FeatureOutputBindingError("feature definition resolution does not match its pinned RSI identity") - feed_identity = "|".join( - ( - "feature-output-feed", - definition_hash, - calculator_version, - resolution, - FEATURE_SERIES_SCHEMA_VERSION, - ) - ) - expected = str(uuid.uuid5(PROJECT_UUID_NAMESPACE, feed_identity)) - if expected != expected_feed_id: - raise FeatureOutputBindingError("feature feed identity inputs do not match the official RSI adapter") return expected_feed_id @@ -250,8 +262,8 @@ def _require_metadata( period_start = _utc(record.get("period_start"), "period_start") period_end = _utc(record.get("period_end"), "period_end") - if period_start > evaluation_from - requirement.warmup_span: - raise FeatureOutputBindingError("feature period does not cover the required warm-up") + if period_start > evaluation_from: + raise FeatureOutputBindingError("feature period does not cover the evaluation start") if period_end < evaluation_through: raise FeatureOutputBindingError("feature period does not cover the full evaluation window") diff --git a/src/backtest_engine/result_snapshot.py b/src/backtest_engine/result_snapshot.py index be943b6..8885315 100644 --- a/src/backtest_engine/result_snapshot.py +++ b/src/backtest_engine/result_snapshot.py @@ -1218,7 +1218,7 @@ def _fill_ledger( fill_count = closing_count = winning_count = losing_count = 0 realized_pnl = total_fees = total_slippage = ZERO - for record in _causal_fill_order(ordered): + for record in _causal_fill_order(run_snapshot.initial_cash, ordered): quantity = record.quantity gross_amount = record.gross_amount fee = record.fee @@ -1325,55 +1325,67 @@ def _fill_ledger( def _causal_fill_order( + initial_cash: Decimal, ordered: tuple[ResultRecord, ...], ) -> tuple[ResultRecord, ...]: - """Restore the causal order of fills sharing one market-data instant. + """Restore causal fill order from successive immutable position snapshots. Record ids are content-derived and therefore cannot encode the order in - which multiple orders consumed the same bar. Their successive position - snapshots do encode that order, so follow that chain before rebuilding the - ledger. + which multiple orders consumed overlapping mixed-resolution bars. Their + bar-end timestamps can also differ from consumption order: a coarser bar + becomes available after finer bars whose timestamps overlap it. Successive + position snapshots do encode the actual order, so follow that chain across + the complete fill set before rebuilding the ledger. """ fills = [item for item in ordered if item.kind is ResultRecordKind.FILL] + pending_ids = {item.record_id for item in fills} + by_predecessor_cash: dict[Decimal, list[ResultRecord]] = {} + for record in fills: + if record.gross_amount is None or record.fee is None: # pragma: no cover + continue + buy_before = quantize_money( + record.cash_after + record.gross_amount + record.fee, "cash_before" + ) + sell_before = quantize_money( + record.cash_after - record.gross_amount + record.fee, "cash_before" + ) + by_predecessor_cash.setdefault(buy_before, []).append(record) + if sell_before != buy_before: + by_predecessor_cash.setdefault(sell_before, []).append(record) result: list[ResultRecord] = [] book: dict[str, PositionAfter] = {} - index = 0 - while index < len(fills): - instant = fills[index].occurred_at - pending: list[ResultRecord] = [] - while index < len(fills) and fills[index].occurred_at == instant: - pending.append(fills[index]) - index += 1 - - while pending: - matched_index = next( - ( - candidate_index - for candidate_index, candidate in enumerate(pending) - if _matches_position_transition(book, candidate) - ), - None, - ) - if matched_index is None: - # Preserve the deterministic record ordering so the canonical - # validation below reports its detailed contradiction. - result.extend(pending) - break - record = pending.pop(matched_index) - result.append(record) - book = { - item.instrument_id: item - for item in record.positions_after - if item.quantity > ZERO - } + cash = quantize_money(initial_cash, "initial_cash") + while pending_ids: + matched_record = next( + ( + candidate + for candidate in by_predecessor_cash.get(cash, ()) + if candidate.record_id in pending_ids + and _matches_fill_transition(book, cash, candidate) + ), + None, + ) + if matched_record is None: + # Preserve deterministic canonical order so ledger validation emits + # the first concrete contradictory transition. + result.extend(item for item in fills if item.record_id in pending_ids) + break + pending_ids.remove(matched_record.record_id) + result.append(matched_record) + book = { + item.instrument_id: item + for item in matched_record.positions_after + if item.quantity > ZERO + } + cash = matched_record.cash_after return tuple(result) -def _matches_position_transition( - book: dict[str, PositionAfter], record: ResultRecord +def _matches_fill_transition( + book: dict[str, PositionAfter], cash: Decimal, record: ResultRecord ) -> bool: - if record.quantity is None: + if record.quantity is None or record.gross_amount is None or record.fee is None: return False after = { item.instrument_id: item @@ -1386,7 +1398,14 @@ def _matches_position_transition( after_quantity = after.get( record.instrument_id, PositionAfter(record.instrument_id, ZERO, ZERO) ).quantity - if (after_quantity - before_quantity).copy_abs() != record.quantity: + delta = after_quantity - before_quantity + if delta == record.quantity: + expected_cash = quantize_money(cash - record.gross_amount - record.fee, "cash_after") + elif delta == -record.quantity: + expected_cash = quantize_money(cash + record.gross_amount - record.fee, "cash_after") + else: + return False + if record.cash_after != expected_cash: return False return all( book.get(instrument_id) == after.get(instrument_id) diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 8ea5087..ac0b5c2 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -1560,6 +1560,27 @@ def _condition_outcomes(evaluation: PlanEvaluation) -> tuple[ConditionOutcome, . # ========================================================================== +def _required_feature_through( + schedule: OfficialSessionSchedule, + evaluation_from: datetime, + evaluation_through: datetime, +) -> datetime: + """Return the last instant at which an evaluation can consume a feature. + + Explicit run dates are converted from the policy timezone to a half-open UTC + interval. Requiring a feature output to cover that wall-clock boundary + incorrectly rejects a complete trading day (for example 20:00 UTC close + versus the following 04:00 UTC New York date boundary). Features are only + consumed by market events, so the final official session close is the + meaningful coverage boundary. + """ + last_close = max( + (session.closes_at for session in schedule.sessions), + default=evaluation_from, + ) + return min(last_close, evaluation_through) + + def dataset_coverage(manifest: Mapping[str, Any]) -> tuple[datetime, datetime]: """The union of the pinned objects' declared coverage, in UTC. @@ -2164,7 +2185,9 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: source=self._feature_materializations, reader=self._feature_object_reader, evaluation_from=evaluation_from, - evaluation_through=evaluation_through, + evaluation_through=_required_feature_through( + evaluation_schedule, evaluation_from, evaluation_through + ), ) except FeatureOutputBindingError as exc: raise JobNotSatisfiable( diff --git a/tests/test_basic_element_conformance.py b/tests/test_basic_element_conformance.py index cb61c29..b259ff7 100644 --- a/tests/test_basic_element_conformance.py +++ b/tests/test_basic_element_conformance.py @@ -8,7 +8,9 @@ import pytest from backtest_engine.elements import ( + ElementCompatibilityError, ElementEvaluation, + ElementEvaluationError, ElementInputMissing, InstrumentInput, PinnedFeatureSeries, @@ -21,6 +23,18 @@ FIXTURE = Path(__file__).parent / "fixtures/contracts/basic-element-conformance.v1.json" INSTRUMENT = "00000000-0000-4000-8000-000000000301" AS_OF = datetime(2026, 8, 26, 20, 0, tzinfo=timezone.utc) +TERMINAL_ARGUMENTS = { + "allocation": "EQUAL", + "orderType": "MARKET", + "timeInForce": "DAY", + "side": "BUY", + "orderPercent": "25", + "maxPositionPercent": "40", + "executionMode": "1회만", + "waitMode": "조건 재충족", + "waitInterval": "1", + "maxExecutions": "1", +} def test_v2_catalog_accepts_every_compiled_operation_and_argument_from_the_corpus() -> None: @@ -39,9 +53,7 @@ def test_v2_catalog_accepts_every_compiled_operation_and_argument_from_the_corpu timeInForce="DAY", side="BUY", ) - catalog.validate_step( - PlanStep(sequence=1, operation=case["operation"], arguments=arguments) - ) + catalog.validate_step(PlanStep(sequence=1, operation=case["operation"], arguments=arguments)) def _values(operation: str, passed: bool) -> dict[str, str]: @@ -73,15 +85,19 @@ def _evaluation(operation: str, passed: bool) -> ElementEvaluation: features = () if operation == "RSI_CROSS": period = timedelta(hours=1) - features = (PinnedFeatureSeries( - feature_id="RSI_14", - instrument_id=INSTRUMENT, - resolution="1h", - values=( - PinnedFeatureValue(AS_OF - period * 2, Decimal("29.00000000") if passed else Decimal("31.00000000")), - PinnedFeatureValue(AS_OF - period, Decimal("31.00000000") if passed else Decimal("32.00000000")), + features = ( + PinnedFeatureSeries( + feature_id="RSI_14", + instrument_id=INSTRUMENT, + resolution="1h", + values=( + PinnedFeatureValue( + AS_OF - period * 2, Decimal("29.00000000") if passed else Decimal("31.00000000") + ), + PinnedFeatureValue(AS_OF - period, Decimal("31.00000000") if passed else Decimal("32.00000000")), + ), ), - ),) + ) return ElementEvaluation( instrument_id=INSTRUMENT, as_of=AS_OF, @@ -135,3 +151,406 @@ def test_missing_history_is_unavailable_and_an_open_bar_waits_instead_of_becomin outcome = catalog.evaluate(price, open_bar) assert outcome.is_passed is False assert outcome.reason_code == "WAITING_FOR_BAR_CLOSE" + + +@pytest.mark.parametrize( + ("operation", "arguments", "invalid_name", "invalid_value"), + [ + ( + "PRICE_CHANGE_PERCENT", + {"resolution": "30m", "base": "PREVIOUS_CLOSE", "direction": "UP", "thresholdPercent": "1"}, + "thresholdPercent", + "-0.00000001", + ), + ( + "RSI_CROSS", + {"resolution": "30m", "direction": "UP", "period": "14", "threshold": "50"}, + "threshold", + "-0.00000001", + ), + ( + "RSI_CROSS", + {"resolution": "30m", "direction": "UP", "period": "14", "threshold": "50"}, + "threshold", + "100.00000001", + ), + ("POSITION_RETURN", {"direction": "PROFIT", "thresholdPercent": "5"}, "thresholdPercent", "100.00000001"), + ("HOLDING_PERIOD", {"unit": "BAR", "amount": "1", "resolution": "30m"}, "amount", "-1"), + ("HOLDING_PERIOD", {"unit": "BAR", "amount": "1", "resolution": "30m"}, "amount", "1.5"), + ("PEAK_RETURN", {"operator": "GTE", "thresholdPercent": "5"}, "thresholdPercent", "-0.00000001"), + ("DRAWDOWN_FROM_PEAK", {"operator": "GTE", "thresholdPercent": "5"}, "thresholdPercent", "100.00000001"), + ("SCHEDULE", {"cycle": "EVERY_N_TRADING_DAYS", "interval": "1", "resolution": "1d"}, "interval", "0"), + ("SCHEDULE", {"cycle": "EVERY_N_TRADING_DAYS", "interval": "1", "resolution": "1d"}, "interval", "1.5"), + ("EMIT_ORDER_CANDIDATE", TERMINAL_ARGUMENTS.copy(), "orderPercent", "0"), + ("EMIT_ORDER_CANDIDATE", TERMINAL_ARGUMENTS.copy(), "maxPositionPercent", "100.00000001"), + ("EMIT_ORDER_CANDIDATE", TERMINAL_ARGUMENTS.copy(), "waitInterval", "0"), + ("EMIT_ORDER_CANDIDATE", TERMINAL_ARGUMENTS.copy(), "maxExecutions", "1.5"), + ], +) +def test_v2_plan_loader_rejects_every_numeric_value_immediately_outside_the_published_boundary( + operation: str, + arguments: dict[str, str], + invalid_name: str, + invalid_value: str, +) -> None: + arguments[invalid_name] = invalid_value + + with pytest.raises(ElementCompatibilityError, match=invalid_name): + element_catalog("basic-elements:2026-08-25").validate_step( + PlanStep(sequence=1, operation=operation, arguments=arguments) + ) + + +@pytest.mark.parametrize("malformed", ["not-a-number", "NaN", "Infinity", "-Infinity"]) +def test_malformed_or_non_finite_market_values_fail_with_a_typed_runtime_error(malformed: str) -> None: + step = PlanStep( + sequence=1, + operation="PRICE_COMPARE", + arguments={"resolution": "30m", "operator": "GT", "reference": "PREVIOUS_CLOSE"}, + ) + evaluation = ElementEvaluation( + instrument_id=INSTRUMENT, + as_of=AS_OF, + inputs=InstrumentInput( + instrument_id=INSTRUMENT, + series=(), + values={"bar.closed.30m": "true", "closes.30m": f"100,{malformed}"}, + ), + ) + + with pytest.raises(ElementEvaluationError, match="PRICE_COMPARE.*closes.30m"): + element_catalog("basic-elements:2026-08-25").evaluate(step, evaluation) + + +def _evaluate_values(operation: str, arguments: dict[str, str], values: dict[str, str]) -> ElementEvaluation: + resolution = arguments.get("resolution") + runtime_values = dict(values) + if resolution and operation not in {"HOLDING_PERIOD", "SCHEDULE"}: + runtime_values[f"bar.closed.{resolution}"] = "true" + return ElementEvaluation( + instrument_id=INSTRUMENT, + as_of=AS_OF, + inputs=InstrumentInput(instrument_id=INSTRUMENT, series=(), values=runtime_values), + ) + + +@pytest.mark.parametrize( + ("operator", "left", "right", "expected"), + [ + ("LT", "99", "100", True), + ("LT", "100", "100", False), + ("LTE", "100", "100", True), + ("LTE", "101", "100", False), + ("GT", "101", "100", True), + ("GT", "100", "100", False), + ("GTE", "100", "100", True), + ("GTE", "99", "100", False), + ("EQ", "100.00000000", "100", True), + ("EQ", "99.99999999", "100", False), + ("NEQ", "99.99999999", "100", True), + ("NEQ", "100", "100.00000000", False), + ], +) +def test_price_comparison_exercises_all_operators_at_and_immediately_around_equality( + operator: str, left: str, right: str, expected: bool +) -> None: + step = PlanStep( + sequence=1, + operation="PRICE_COMPARE", + arguments={"resolution": "30m", "operator": operator, "reference": "PREVIOUS_CLOSE"}, + ) + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + step, _evaluate_values(step.operation, dict(step.arguments), {"closes.30m": f"{right},{left}"}) + ) + + assert outcome.is_passed is expected + + +@pytest.mark.parametrize("resolution", ["30m", "1h", "4h", "1d"]) +def test_each_published_resolution_reads_only_its_own_series(resolution: str) -> None: + step = PlanStep( + sequence=1, + operation="PRICE_COMPARE", + arguments={"resolution": resolution, "operator": "GT", "reference": "PREVIOUS_CLOSE"}, + ) + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + step, + _evaluate_values(step.operation, dict(step.arguments), {f"closes.{resolution}": "100,101"}), + ) + + assert outcome.is_passed is True + + +@pytest.mark.parametrize( + ("reference", "operator", "closes", "extra"), + [ + ("PREVIOUS_CLOSE", "GT", "100,101", {}), + ("SESSION_OPEN", "GT", "101", {"session.open": "100"}), + ("AVERAGE_ENTRY_PRICE", "GT", "101", {"position.averageEntryPrice": "100"}), + ("SMA_5", "GT", ",".join(["100"] * 4 + ["110"]), {}), + ("SMA_20", "GT", ",".join(["100"] * 19 + ["110"]), {}), + ("SMA_60", "GT", ",".join(["100"] * 59 + ["110"]), {}), + ("HIGH_5", "GT", ",".join(["100"] * 5 + ["110"]), {}), + ("HIGH_20", "GT", ",".join(["100"] * 20 + ["110"]), {}), + ("HIGH_60", "GT", ",".join(["100"] * 60 + ["110"]), {}), + ("LOW_5", "LT", ",".join(["100"] * 5 + ["90"]), {}), + ("LOW_20", "LT", ",".join(["100"] * 20 + ["90"]), {}), + ("LOW_60", "LT", ",".join(["100"] * 60 + ["90"]), {}), + ], +) +def test_every_price_reference_uses_its_declared_source( + reference: str, operator: str, closes: str, extra: dict[str, str] +) -> None: + step = PlanStep( + sequence=1, + operation="PRICE_COMPARE", + arguments={"resolution": "30m", "operator": operator, "reference": reference}, + ) + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + step, + _evaluate_values(step.operation, dict(step.arguments), {"closes.30m": closes, **extra}), + ) + + assert outcome.is_passed is True + + +@pytest.mark.parametrize( + ("operation", "direction", "arguments", "values", "features"), + [ + ( + "PRICE_CHANGE_PERCENT", + "UP", + {"resolution": "30m", "base": "PREVIOUS_CLOSE", "thresholdPercent": "5"}, + {"closes.30m": "100,105"}, + (), + ), + ( + "PRICE_CHANGE_PERCENT", + "DOWN", + {"resolution": "30m", "base": "PREVIOUS_CLOSE", "thresholdPercent": "5"}, + {"closes.30m": "100,95"}, + (), + ), + ("STREAK", "UP", {"resolution": "1d", "bars": "3"}, {"closes.1d": "100,101,102,103"}, ()), + ("STREAK", "DOWN", {"resolution": "1d", "bars": "3"}, {"closes.1d": "103,102,101,100"}, ()), + ( + "SMA_CROSS", + "UP", + {"resolution": "30m", "shortPeriod": "5", "longPeriod": "20"}, + {"closes.30m": ",".join(["100"] * 16 + ["90"] * 4 + ["200"])}, + (), + ), + ( + "SMA_CROSS", + "DOWN", + {"resolution": "30m", "shortPeriod": "5", "longPeriod": "20"}, + {"closes.30m": ",".join(["100"] * 16 + ["110"] * 4 + ["0"])}, + (), + ), + ( + "MACD_CROSS", + "UP", + {"resolution": "4h", "fastPeriod": "12", "slowPeriod": "26", "signalPeriod": "9"}, + {"closes.4h": ",".join(["100"] * 37 + ["110"])}, + (), + ), + ( + "MACD_CROSS", + "DOWN", + {"resolution": "4h", "fastPeriod": "12", "slowPeriod": "26", "signalPeriod": "9"}, + {"closes.4h": ",".join(["100"] * 37 + ["90"])}, + (), + ), + ( + "BOLLINGER_REVERSAL", + "UP", + {"resolution": "1d", "period": "20", "deviations": "2"}, + {"closes.1d": ",".join(["100"] * 19 + ["80", "100"])}, + (), + ), + ( + "BOLLINGER_REVERSAL", + "DOWN", + {"resolution": "1d", "period": "20", "deviations": "2"}, + {"closes.1d": ",".join(["100"] * 19 + ["120", "100"])}, + (), + ), + ( + "RSI_CROSS", + "UP", + {"resolution": "1h", "period": "14", "threshold": "30"}, + {}, + ("29", "31"), + ), + ( + "RSI_CROSS", + "DOWN", + {"resolution": "1h", "period": "14", "threshold": "70"}, + {}, + ("71", "69"), + ), + ], +) +def test_every_directional_condition_triggers_on_the_exact_declared_transition( + operation: str, + direction: str, + arguments: dict[str, str], + values: dict[str, str], + features: tuple[str, ...], +) -> None: + arguments = {**arguments, "direction": direction} + evaluation = _evaluate_values(operation, arguments, values) + if features: + period = timedelta(hours=1) + evaluation = ElementEvaluation( + instrument_id=INSTRUMENT, + as_of=AS_OF, + inputs=InstrumentInput( + instrument_id=INSTRUMENT, + series=(), + feature_series=( + PinnedFeatureSeries( + feature_id="RSI_14", + instrument_id=INSTRUMENT, + resolution="1h", + values=( + PinnedFeatureValue(AS_OF - period * 2, Decimal(f"{features[0]}.00000000")), + PinnedFeatureValue(AS_OF - period, Decimal(f"{features[1]}.00000000")), + ), + ), + ), + require_pinned_features=True, + values={"bar.closed.1h": "true"}, + ), + ) + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + PlanStep(sequence=1, operation=operation, arguments=arguments), evaluation + ) + + assert outcome.is_passed is True + + +@pytest.mark.parametrize( + ("cycle", "values"), + [ + ("EVERY_TRADING_DAY", {"schedule.newTradingDay": "true"}), + ("WEEK_FIRST_TRADING_DAY", {"schedule.weekFirstTradingDay": "true"}), + ("MONTH_FIRST_TRADING_DAY", {"schedule.monthFirstTradingDay": "true"}), + ("MONTH_LAST_TRADING_DAY", {"schedule.monthLastTradingDay": "true"}), + ("EVERY_N_TRADING_DAYS", {"schedule.newTradingDay": "true", "schedule.tradingDayIndex": "6"}), + ], +) +def test_every_schedule_cycle_has_a_real_true_runtime_path(cycle: str, values: dict[str, str]) -> None: + arguments = {"cycle": cycle, "interval": "5", "resolution": "1d"} + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + PlanStep(sequence=1, operation="SCHEDULE", arguments=arguments), + _evaluate_values("SCHEDULE", arguments, values), + ) + + assert outcome.is_passed is True + + +@pytest.mark.parametrize( + ("unit", "amount", "resolution", "values"), + [ + ("SESSION_CLOSE", "0", "1d", {"session.close": "true"}), + ("BAR", "5", "4h", {"position.holdingBars.4h": "5"}), + ("TRADING_DAY", "5", "1d", {"position.holdingTradingDays": "5"}), + ], +) +def test_every_holding_period_unit_passes_at_its_exact_boundary( + unit: str, amount: str, resolution: str, values: dict[str, str] +) -> None: + arguments = {"unit": unit, "amount": amount, "resolution": resolution} + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + PlanStep(sequence=1, operation="HOLDING_PERIOD", arguments=arguments), + _evaluate_values("HOLDING_PERIOD", arguments, values), + ) + + assert outcome.is_passed is True + + +@pytest.mark.parametrize("period", ["1", "5", "20", "60"]) +@pytest.mark.parametrize("multiplier", ["1", "2", "3"]) +@pytest.mark.parametrize("reference", ["PREVIOUS_VOLUME", "AVERAGE_VOLUME"]) +def test_every_volume_reference_period_and_multiplier_combination_uses_the_declared_window( + period: str, multiplier: str, reference: str +) -> None: + count = int(period) + previous = ["100"] * max(count, 1) + current = str(100 * int(multiplier)) + arguments = { + "resolution": "4h", + "operator": "GTE", + "reference": reference, + "period": period, + "multiplier": multiplier, + } + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + PlanStep(sequence=1, operation="VOLUME_COMPARE", arguments=arguments), + _evaluate_values( + "VOLUME_COMPARE", + arguments, + {"volumes.4h": ",".join([*previous, current])}, + ), + ) + + assert outcome.is_passed is True + + +@pytest.mark.parametrize( + ("operation", "operator", "value", "threshold", "expected"), + [ + ("PEAK_RETURN", "LT", "9.99999999", "10", True), + ("PEAK_RETURN", "LTE", "10", "10", True), + ("PEAK_RETURN", "GT", "10.00000001", "10", True), + ("PEAK_RETURN", "GTE", "10", "10", True), + ("PEAK_RETURN", "EQ", "10.00000000", "10", True), + ("PEAK_RETURN", "NEQ", "9.99999999", "10", True), + ("DRAWDOWN_FROM_PEAK", "LT", "9.99999999", "10", True), + ("DRAWDOWN_FROM_PEAK", "LTE", "10", "10", True), + ("DRAWDOWN_FROM_PEAK", "GT", "10.00000001", "10", True), + ("DRAWDOWN_FROM_PEAK", "GTE", "10", "10", True), + ("DRAWDOWN_FROM_PEAK", "EQ", "10.00000000", "10", True), + ("DRAWDOWN_FROM_PEAK", "NEQ", "9.99999999", "10", True), + ], +) +def test_position_threshold_conditions_cover_all_operators_at_decimal_precision( + operation: str, operator: str, value: str, threshold: str, expected: bool +) -> None: + key = "position.peakReturnPercent" if operation == "PEAK_RETURN" else "position.drawdownPercent" + arguments = {"operator": operator, "thresholdPercent": threshold} + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + PlanStep(sequence=1, operation=operation, arguments=arguments), + _evaluate_values(operation, arguments, {key: value}), + ) + + assert outcome.is_passed is expected + + +@pytest.mark.parametrize( + ("direction", "value", "threshold", "expected"), + [ + ("PROFIT", "5", "5", True), + ("PROFIT", "4.99999999", "5", False), + ("LOSS", "-5", "5", True), + ("LOSS", "-4.99999999", "5", False), + ], +) +def test_position_return_honors_profit_and_loss_boundaries( + direction: str, value: str, threshold: str, expected: bool +) -> None: + arguments = {"direction": direction, "thresholdPercent": threshold} + + outcome = element_catalog("basic-elements:2026-08-25").evaluate( + PlanStep(sequence=1, operation="POSITION_RETURN", arguments=arguments), + _evaluate_values("POSITION_RETURN", arguments, {"position.returnPercent": value}), + ) + + assert outcome.is_passed is expected diff --git a/tests/test_basic_runtime.py b/tests/test_basic_runtime.py index 138d3f6..a1b9c7d 100644 --- a/tests/test_basic_runtime.py +++ b/tests/test_basic_runtime.py @@ -310,19 +310,23 @@ def v2_catalog(document: dict[str, Any]) -> None: @pytest.mark.parametrize( - ("resolution", "wire_resolution", "feature_id"), + ("catalog_version", "resolution", "wire_resolution", "feature_id"), [ - ("30m", "PT30M", "4b1c6801-0259-5176-a857-0e5ea923d898"), - ("1h", "PT1H", "2e18c093-5d4e-5d9a-bd22-b7e5679f1a3e"), - ("4h", "PT4H", "1b2785bd-20f0-50a2-ae96-6a1f7bad74b9"), - ("1d", "PT24H", "eddfb2d4-8586-5260-8fc9-9c8125990270"), + ("basic-elements:2026-08-08", "30m", "PT30M", "4b1c6801-0259-5176-a857-0e5ea923d898"), + ("basic-elements:2026-08-08", "1h", "PT1H", "2e18c093-5d4e-5d9a-bd22-b7e5679f1a3e"), + ("basic-elements:2026-08-08", "4h", "PT4H", "1b2785bd-20f0-50a2-ae96-6a1f7bad74b9"), + ("basic-elements:2026-08-08", "1d", "PT24H", "eddfb2d4-8586-5260-8fc9-9c8125990270"), + ("basic-elements:2026-08-25", "30m", "PT30M", "ec37984b-6605-5560-8ea0-774c5b8e9626"), + ("basic-elements:2026-08-25", "1h", "PT1H", "85f4f80f-be4e-d9dc-bd52-d4781ba5f30f"), + ("basic-elements:2026-08-25", "4h", "PT4H", "65a5aaf5-f536-820f-119a-239b0aec0de7"), + ("basic-elements:2026-08-25", "1d", "PT24H", "647a5fd6-98ed-0617-d4b2-844748d54fac"), ], ) def test_rsi_cross_requires_the_exact_selected_resolution_feature_definition( - resolution: str, wire_resolution: str, feature_id: str + catalog_version: str, resolution: str, wire_resolution: str, feature_id: str ) -> None: def rsi_cross(document: dict[str, Any]) -> None: - catalog = element_catalog("basic-elements:2026-08-08") + catalog = element_catalog(catalog_version) terminal = catalog.spec("EMIT_ORDER_CANDIDATE") terminal_arguments = { name: values[0] for name, values in terminal.enumerations.items() @@ -330,6 +334,8 @@ def rsi_cross(document: dict[str, Any]) -> None: terminal_arguments.update( {"orderPercent": "50", "waitInterval": "1", "maxExecutions": "1"} ) + if "maxPositionPercent" in terminal.required_arguments: + terminal_arguments["maxPositionPercent"] = "100" document["elementCatalogVersion"] = catalog.version document["requiredFeatures"] = [ { diff --git a/tests/test_feature_outputs.py b/tests/test_feature_outputs.py index bd0e314..b5d36ce 100644 --- a/tests/test_feature_outputs.py +++ b/tests/test_feature_outputs.py @@ -77,14 +77,17 @@ def _utc_text(value: datetime) -> str: def _canonical_hash( - rows: list[dict[str, str]], *, definition_hash: str = DEFINITION_HASH + rows: list[dict[str, str]], + *, + definition_hash: str = DEFINITION_HASH, + period_start: datetime = PERIOD_START, ) -> str: payload = { "definition_hash": definition_hash.removeprefix("sha256:"), "input_dataset_set_hash": INPUT_HASH, "instrument_id": INSTRUMENT_ID, "period_end": _utc_text(PERIOD_END), - "period_start": _utc_text(PERIOD_START), + "period_start": _utc_text(period_start), "result_schema_version": 1, "rows": rows, } @@ -353,6 +356,30 @@ def test_proposed_rsi_seed_reproduces_the_exact_deterministic_feed_identity() -> "1d", "6d2647f8-5caf-55ee-8821-869dc693f68a", ), + ( + "ec37984b-6605-5560-8ea0-774c5b8e9626", + "sha256:250df12e46d233e7b8ece86c64df7a3941f0d70436aebe522b1387f15fb346dc", + "30m", + "57794d8c-2254-53e4-966e-44f97edd9e6a", + ), + ( + "85f4f80f-be4e-d9dc-bd52-d4781ba5f30f", + "sha256:7e8c5600ff2bf07a043f797a50d6467f86fbdb56ee532c87929df97f246af2de", + "1h", + "28012549-4f45-56d3-8bb6-329e4c7a9d77", + ), + ( + "65a5aaf5-f536-820f-119a-239b0aec0de7", + "sha256:42e28b02a1552eb2aa42e0d89b1ea3dd909ee8d34c3bc290c4ce0234c6d705da", + "4h", + "e1d7d508-aaf1-5ae9-8098-c4af870f6fa4", + ), + ( + "647a5fd6-98ed-0617-d4b2-844748d54fac", + "sha256:64dbbcda7352d0add9a4a6a6ed94a780603880891684dc32cf39e0a3d1167422", + "1d", + "6d2647f8-5caf-55ee-8821-869dc693f68a", + ), ], ) def test_each_production_resolution_has_one_deterministic_rsi_feed_identity( @@ -441,7 +468,7 @@ def test_exact_pin_is_read_from_the_named_object_version_and_decoded() -> None: (lambda record: record.update(status="FAILED"), "SUCCEEDED"), (lambda record: record.update(calculator_version="rsi:2.0.0"), "semantic version"), (lambda record: record.update(resolution="15m"), "resolution"), - (lambda record: record.update(period_start=PERIOD_START + BAR), "warm-up"), + (lambda record: record.update(period_start=EVALUATION_FROM + BAR), "evaluation start"), (lambda record: record.update(period_end=PERIOD_END - BAR), "evaluation window"), (lambda record: record.update(output_dataset_status="BUILDING"), "AVAILABLE"), (lambda record: record.update(output_dataset_layer="RAW"), "DERIVED"), @@ -476,6 +503,16 @@ def test_missing_pin_fails_the_required_feature_instrument_join() -> None: ) +def test_feature_series_may_warm_up_inside_the_evaluation_period() -> None: + body = _parquet() + result_hash = _canonical_hash(_rows(), period_start=EVALUATION_FROM) + record = _record(body, period_start=EVALUATION_FROM, result_hash=result_hash) + + resolved, _reader = _resolve(records={MATERIALIZATION_ID: record}, body=body) + + assert resolved[0].value_at(EVALUATION_FROM) == Decimal("0.00000000") + + def test_duplicate_tuple_fails_the_required_feature_instrument_join() -> None: body = _parquet() second_id = uuid.UUID("10000000-0000-4000-8000-000000000002") diff --git a/tests/test_result_snapshot.py b/tests/test_result_snapshot.py index df92a67..b7afced 100644 --- a/tests/test_result_snapshot.py +++ b/tests/test_result_snapshot.py @@ -535,6 +535,53 @@ def test_a_fill_whose_position_delta_contradicts_its_quantity_is_rejected() -> N ) +def test_fill_ledger_recovers_causal_position_order_across_overlapping_bar_times() -> None: + """Mixed resolutions can be consumed in an order unlike their bar-end timestamps.""" + opens_four = _record( + kind=ResultRecordKind.FILL, + record_id="00000000-0000-4000-8000-000000000851", + occurred_at="2025-11-28T16:00:00Z", + order_id=ORDER_ID, + status=OrderStatus.FILLED, + cash_after="9600", + positions_after=(_position(quantity="4", cost_basis="400"),), + fill_id="00000000-0000-4000-8000-000000000852", + quantity="4", + base_price="100", + price="100", + gross_amount="400", + slippage_amount="0", + fee="0", + cost_basis="400", + realized_pnl="0", + ) + adds_twenty_seven = _record( + kind=ResultRecordKind.FILL, + record_id="00000000-0000-4000-8000-000000000853", + occurred_at="2025-11-28T15:00:00Z", + order_id=OTHER_ORDER_ID, + status=OrderStatus.FILLED, + cash_after="6900", + positions_after=(_position(quantity="31", cost_basis="3100"),), + fill_id="00000000-0000-4000-8000-000000000854", + quantity="27", + base_price="100", + price="100", + gross_amount="2700", + slippage_amount="0", + fee="0", + cost_basis="2700", + realized_pnl="0", + ) + + result = ResultSnapshotBuilder().build( + _run(), [adds_twenty_seven, opens_four], _utc("2025-11-28T17:00:00Z") + ) + + assert result.summary.ending_positions == (_position(quantity="31", cost_basis="3100"),) + assert result.summary.metrics["fillCount"].value == Decimal("2") + + def _mark_series() -> ValuationSeries: return ValuationSeries( basis=ValuationBasis.MARK_TO_MARKET, diff --git a/tests/test_wiring.py b/tests/test_wiring.py index 8905fe7..031bf83 100644 --- a/tests/test_wiring.py +++ b/tests/test_wiring.py @@ -49,7 +49,7 @@ ) from backtest_engine.elements import SeriesBar, resolution_period from backtest_engine.elements.orders import OrderCandidate -from backtest_engine.event_clock import MarketDataEvent, MarketEventClock +from backtest_engine.event_clock import MarketDataEvent, MarketEventClock, OfficialSessionSchedule from backtest_engine.execution_model import BacktestExecutionModel, RiskLimits from backtest_engine.orchestrator import ( PlanReplay, @@ -70,6 +70,7 @@ _CancellationAwareMonitor, _dataset_cover_contains_evaluation, _metric_percent, + _required_feature_through, _resolve_position_only_flow_clocks, dataset_coverage, evaluation_window, @@ -97,6 +98,20 @@ INITIAL_CASH = Decimal("100000.00000000") +def test_feature_coverage_ends_at_the_last_market_close_not_the_local_date_boundary() -> None: + evaluation_from = datetime(2026, 7, 29, 4, 0, tzinfo=UTC) + evaluation_through = datetime(2026, 7, 30, 4, 0, tzinfo=UTC) + schedule = XNYS_CALENDAR.session_schedule(date(2026, 7, 29), date(2026, 7, 29)) + + assert _required_feature_through(schedule, evaluation_from, evaluation_through) == datetime( + 2026, 7, 29, 20, 0, tzinfo=UTC + ) + empty = OfficialSessionSchedule(date(2026, 7, 30), date(2026, 7, 30), ()) + assert _required_feature_through(empty, evaluation_from, evaluation_through) == evaluation_from + partial_session_end = datetime(2026, 7, 29, 16, 0, tzinfo=UTC) + assert _required_feature_through(schedule, evaluation_from, partial_session_end) == partial_session_end + + def _plan() -> Any: return BasicPlanRuntime().load(compiled_plan()) From d928d95d72d9bfecc1e61aaf2322a0f754fa4018 Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Fri, 28 Aug 2026 13:29:05 +0900 Subject: [PATCH 8/9] style: format backtest persistence repository --- .../persistence/repositories.py | 24 +++++++------------ 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/src/backtest_engine/persistence/repositories.py b/src/backtest_engine/persistence/repositories.py index 0080333..0fc0cc9 100644 --- a/src/backtest_engine/persistence/repositories.py +++ b/src/backtest_engine/persistence/repositories.py @@ -345,9 +345,7 @@ def request_cancellation(self, run_id: UUID, *, reason_code: str) -> RunRow: def request_deletion(self, run_id: UUID) -> RunRow: """Cancel active work and retain the row as immutable owner evidence.""" - current = self._connection.execute( - select(runs).where(runs.c.id == run_id).with_for_update() - ).mappings().first() + current = self._connection.execute(select(runs).where(runs.c.id == run_id).with_for_update()).mappings().first() if current is None: raise RowNotFound(f"backtest run not found: {run_id}") hydrated = _hydrate(RunRow, current) @@ -379,9 +377,11 @@ def request_deletion(self, run_id: UUID) -> RunRow: cancellation_requested_at=hydrated.cancellation_requested_at or requested_at, cancellation_reason_code=hydrated.cancellation_reason_code or "USER_DELETED", ) - updated = self._connection.execute( - update(runs).where(runs.c.id == run_id).values(**values).returning(*runs.c) - ).mappings().one() + updated = ( + self._connection.execute(update(runs).where(runs.c.id == run_id).values(**values).returning(*runs.c)) + .mappings() + .one() + ) return _hydrate(RunRow, updated) def _transition(self, run_id: UUID, target: RunStatus, **values: Any) -> RunRow: @@ -593,9 +593,7 @@ def release_fenced( ) -> RunAttemptRow: """Close exactly one live delivery and make its non-terminal run retryable.""" attempt = ( - self._connection.execute( - select(run_attempts.c.run_id).where(run_attempts.c.id == attempt_id) - ) + self._connection.execute(select(run_attempts.c.run_id).where(run_attempts.c.id == attempt_id)) .mappings() .first() ) @@ -633,9 +631,7 @@ def close_fenced( if status in (WorkStatus.PENDING, WorkStatus.RUNNING): raise ValueError("close_fenced requires a terminal status") attempt = ( - self._connection.execute( - select(run_attempts.c.run_id).where(run_attempts.c.id == attempt_id) - ) + self._connection.execute(select(run_attempts.c.run_id).where(run_attempts.c.id == attempt_id)) .mappings() .first() ) @@ -653,9 +649,7 @@ def close_fenced( if run is None: raise RowNotFound(f"backtest run not found: {attempt['run_id']}") locked_attempt = self._connection.execute( - select(run_attempts.c.id) - .where(run_attempts.c.id == attempt_id) - .with_for_update() + select(run_attempts.c.id).where(run_attempts.c.id == attempt_id).with_for_update() ).first() if locked_attempt is None: raise StaleAttemptClaim("terminal mutation matched no attempt") From b51a4f417bdb5b6be8baeae6e265796f1a60e02c Mon Sep 17 00:00:00 2001 From: Ryan Min Date: Fri, 28 Aug 2026 13:39:06 +0900 Subject: [PATCH 9/9] fix: close backtest integration boundaries --- src/backtest_engine/orchestrator.py | 2 +- src/backtest_engine/persistence/repositories.py | 8 +++++--- tests/persistence/test_roundtrip.py | 9 +++++++-- tests/test_reproducibility_e2e.py | 13 +++++++------ 4 files changed, 20 insertions(+), 12 deletions(-) diff --git a/src/backtest_engine/orchestrator.py b/src/backtest_engine/orchestrator.py index 799846a..4987f17 100644 --- a/src/backtest_engine/orchestrator.py +++ b/src/backtest_engine/orchestrator.py @@ -799,7 +799,7 @@ def run( event for event in combined_events if event.occurred_at >= min(requirement.warmup_from for requirement in job.requirements) - and (job.evaluation_through is None or event.occurred_at < job.evaluation_through) + and (job.evaluation_through is None or event.occurred_at <= job.evaluation_through) ), start=1, ) diff --git a/src/backtest_engine/persistence/repositories.py b/src/backtest_engine/persistence/repositories.py index 0fc0cc9..d67159f 100644 --- a/src/backtest_engine/persistence/repositories.py +++ b/src/backtest_engine/persistence/repositories.py @@ -304,13 +304,15 @@ def mark_cancelled( reason_code: str, ) -> RunRow: current = self.get(run_id) + requested_at = current.cancellation_requested_at or cancelled_at + effective_cancelled_at = max(cancelled_at, requested_at) return self._transition( run_id, RunStatus.CANCELLED, - completed_at=cancelled_at, - cancellation_requested_at=current.cancellation_requested_at or cancelled_at, + completed_at=effective_cancelled_at, + cancellation_requested_at=requested_at, cancellation_reason_code=current.cancellation_reason_code or reason_code, - cancelled_at=cancelled_at, + cancelled_at=effective_cancelled_at, ) def request_cancellation(self, run_id: UUID, *, reason_code: str) -> RunRow: diff --git a/tests/persistence/test_roundtrip.py b/tests/persistence/test_roundtrip.py index 957801b..453d163 100644 --- a/tests/persistence/test_roundtrip.py +++ b/tests/persistence/test_roundtrip.py @@ -282,11 +282,16 @@ def test_owner_soft_delete_waits_for_running_worker_then_hides_terminal_evidence cancelled = uow.runs.mark_cancelled(run.id, cancelled_at, "USER_DELETED") assert cancelled.status is RunStatus.CANCELLED - assert cancelled.deleted_at == max(cancelled_at, pending.deletion_requested_at) + effective_cancelled_at = max(cancelled_at, pending.deletion_requested_at) + assert cancelled.cancelled_at == effective_cancelled_at + assert cancelled.completed_at == effective_cancelled_at + assert cancelled.deleted_at == effective_cancelled_at with persistence.unit_of_work() as uow: with pytest.raises(RowNotFound): uow.runs.get_owned(ACCOUNT_ID, run.id) - assert uow.runs.get(run.id).deleted_at == cancelled_at + persisted = uow.runs.get(run.id) + assert persisted.cancelled_at == effective_cancelled_at + assert persisted.deleted_at == effective_cancelled_at def test_lifecycle_transitions_use_the_canonical_completed_label( diff --git a/tests/test_reproducibility_e2e.py b/tests/test_reproducibility_e2e.py index c82826f..1f6b6ee 100644 --- a/tests/test_reproducibility_e2e.py +++ b/tests/test_reproducibility_e2e.py @@ -116,11 +116,11 @@ EXPECTED_RUN_SNAPSHOT_ID = "75fd83d0c9cd6356a9c0ed1db9833881f19a0a136042bf96d82617085ba64348" #: `backtest.performance_summaries.result_hash`. -EXPECTED_RESULT_HASH = "ce53f523451e506c2ec8264043b9221729eefe8c2040088883a30dc922d56c08" +EXPECTED_RESULT_HASH = "d32135e1775553edc037d17fea01afc1e04f30844653741b77b235cf3677470b" #: `storage.objects.content_hash` of the TRADE_DETAIL Parquet part. EXPECTED_TRADE_DETAIL_CONTENT_HASH = ( - "28977f43a1a3cb811538affd6a2a98903641060089b24e2de48d84d925be5629" + "054779480a7c8f4cd4a1d16cebd67455fe141529ac242f57cec81039031d9188" ) @@ -235,8 +235,9 @@ def test_an_official_request_traverses_http_sqs_worker_postgres_and_the_object_s id=run_id, ) assert [item["et_year_month"] for item in monthly] == ["2024-01"] - # 20 one-minute bars -> 20 evaluation instants; exactly one of them decided. - assert monthly[0]["evaluation_count"] == len(CLOSES) + # RSI_14 needs 15 prior closes, so only bars 15..19 are executable + # evaluation instants; exactly one of those five instants decided. + assert monthly[0]["evaluation_count"] == len(CLOSES) - 15 assert monthly[0]["triggered_count"] == 1 assert monthly[0]["data_gap_count"] == 0 assert monthly[0]["trade_event_count"] == 2 # the accepted order and its fill @@ -666,7 +667,7 @@ def test_a_dataset_too_short_for_the_warmup_fails_the_run_and_dead_letters_the_j handled = stack.worker.poll_once() assert [item.disposition for item in handled] == [MessageDisposition.DEAD_LETTERED] - assert [item.reason_code for item in handled] == ["REQUIRED_DATA_UNAVAILABLE"] + assert [item.reason_code for item in handled] == ["REQUIRED_INPUT_UNAVAILABLE"] assert stack.visible(stack.dead_letter_queue) == 1 row = sql_one( @@ -675,7 +676,7 @@ def test_a_dataset_too_short_for_the_warmup_fails_the_run_and_dead_letters_the_j id=run_id, ) assert row["status"] == "FAILED" - assert row["failure_code"] == "REQUIRED_DATA_UNAVAILABLE" + assert row["failure_code"] == "REQUIRED_INPUT_UNAVAILABLE" assert sql_all( admin_engine, "SELECT run_id FROM backtest.performance_summaries WHERE run_id = :id", id=run_id ) == []