From 0a0f69cdc6f9f1720100c69df6bb1309231c1254 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 5 Aug 2026 02:41:27 +0800 Subject: [PATCH] feat(backtest): enforce promotion purged WFA Co-Authored-By: Codex --- .../backtest_orchestrator.py | 417 +++++++++++++++-- .../strategy_lifecycle/contracts.py | 127 +++++- tests/test_backtest_orchestrator.py | 427 +++++++++++++++++- 3 files changed, 935 insertions(+), 36 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py b/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py index efedb952..350ef485 100644 --- a/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py +++ b/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py @@ -5,16 +5,45 @@ from __future__ import annotations +import calendar import itertools +import math +import re import uuid from dataclasses import replace -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone from typing import Any, Mapping, Protocol, Sequence, runtime_checkable -from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult, SensitivityReport +from quant_platform_kit.strategy_lifecycle.contracts import ( + BacktestResult, + BacktestValidationIdentity, + PromotionBacktestRun, + PromotionCostModel, + PurgedWalkForwardFold, + SensitivityReport, +) from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore +_PROMOTION_PROTOCOL = "purged_walk_forward.v1" +_SOURCE_REVISION_PATTERN = re.compile(r"[0-9a-f]{40}") +_PROMOTION_REQUIRED_METRICS = ("sharpe_ratio", "max_drawdown", "cagr") +_PROMOTION_OPTIONAL_METRICS = ( + "calmar_ratio", + "sortino_ratio", + "volatility", + "win_rate", + "total_return", + "benchmark_cagr", + "benchmark_max_drawdown", + "excess_cagr", + "oos_sharpe", + "oos_calmar", + "oos_max_drawdown", + "walk_forward_stability", +) + + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -23,6 +52,136 @@ def _run_id() -> str: return uuid.uuid4().hex[:12] +def _is_finite_number(value: object) -> bool: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(float(value)) + ) + + +def _add_calendar_months(value: date, months: int) -> date: + month_index = value.month - 1 + months + year = value.year + month_index // 12 + month = month_index % 12 + 1 + day = min(value.day, calendar.monthrange(year, month)[1]) + return date(year, month, day) + + +def _validate_promotion_plan( + folds: Sequence[PurgedWalkForwardFold], + *, + locked_oos_start: date, + locked_oos_end: date, + purge_days: int, + embargo_days: int, + source_revision: str, + cost_model: PromotionCostModel, +) -> tuple[PurgedWalkForwardFold, ...]: + for index, fold in enumerate(folds): + if not isinstance(fold, PurgedWalkForwardFold): + raise TypeError(f"folds[{index}] must be a PurgedWalkForwardFold") + if len(folds) < 3: + raise ValueError( + "promotion-grade orchestration requires at least three Purged Walk-Forward folds" + ) + if ( + not isinstance(purge_days, int) + or isinstance(purge_days, bool) + or purge_days <= 0 + ): + raise ValueError("purge_days must be an explicit positive integer") + if ( + not isinstance(embargo_days, int) + or isinstance(embargo_days, bool) + or embargo_days <= 0 + ): + raise ValueError("embargo_days must be an explicit positive integer") + if type(locked_oos_start) is not date or type(locked_oos_end) is not date: + raise TypeError("locked OOS boundaries must be calendar dates") + if not _SOURCE_REVISION_PATTERN.fullmatch(source_revision): + raise ValueError( + "source_revision must be a lowercase 40-character Git revision" + ) + if not isinstance(cost_model, PromotionCostModel): + raise TypeError("cost_model must be a PromotionCostModel") + if not cost_model.model_id.strip(): + raise ValueError("cost_model.model_id must be non-empty") + for name in ("commission_bps", "slippage_bps", "market_impact_bps"): + value = getattr(cost_model, name) + if not _is_finite_number(value) or float(value) < 0: + raise ValueError(f"cost_model.{name} must be finite and non-negative") + + validated_folds: list[PurgedWalkForwardFold] = [] + previous_test_end: date | None = None + for index, fold in enumerate(folds): + boundaries = (fold.train_start, fold.train_end, fold.test_start, fold.test_end) + if any(type(boundary) is not date for boundary in boundaries): + raise TypeError(f"folds[{index}] boundaries must be calendar dates") + if fold.train_start > fold.train_end: + raise ValueError(f"folds[{index}] train boundaries are reversed") + if fold.test_start > fold.test_end: + raise ValueError(f"folds[{index}] test boundaries are reversed") + if fold.train_end + timedelta(days=purge_days) >= fold.test_start: + raise ValueError( + f"folds[{index}] does not contain the required purge interval" + ) + if ( + previous_test_end is not None + and previous_test_end + timedelta(days=embargo_days) >= fold.train_start + ): + raise ValueError( + f"folds[{index}] overlaps or violates the ordered embargo boundary" + ) + validated_folds.append(fold) + previous_test_end = fold.test_end + + if ( + previous_test_end is None + or previous_test_end + timedelta(days=embargo_days) >= locked_oos_start + ): + raise ValueError( + "locked OOS overlaps the folds or violates the embargo boundary" + ) + if locked_oos_end < _add_calendar_months(locked_oos_start, 12): + raise ValueError("locked OOS must span at least 12 calendar months") + return tuple(validated_folds) + + +def _validate_promotion_result( + result: object, *, start_date: date, end_date: date +) -> BacktestResult: + if not isinstance(result, BacktestResult): + raise TypeError("promotion runner must return BacktestResult") + if result.start_date != start_date or result.end_date != end_date: + raise ValueError( + "promotion result must retain the exact dated window requested by the orchestrator" + ) + if not isinstance(result.observation_count, int) or isinstance( + result.observation_count, bool + ): + raise ValueError( + "promotion result observation_count must be a positive integer" + ) + if result.observation_count <= 0: + raise ValueError("promotion result observation_count must be positive") + for name in _PROMOTION_REQUIRED_METRICS: + if not _is_finite_number(getattr(result, name)): + raise ValueError(f"promotion result {name} must be present and finite") + for name in _PROMOTION_OPTIONAL_METRICS: + value = getattr(result, name) + if value is not None and not _is_finite_number(value): + raise ValueError(f"promotion result {name} must be finite when provided") + if ( + not _is_finite_number(result.run_duration_seconds) + or result.run_duration_seconds < 0 + ): + raise ValueError( + "promotion result run_duration_seconds must be finite and non-negative" + ) + return result + + @runtime_checkable class BacktestRunner(Protocol): """Protocol that each market adapter must implement. @@ -52,6 +211,32 @@ def run( ... +@runtime_checkable +class PromotionBacktestRunner(Protocol): + """Explicit capability contract for promotion-grade backtest adapters.""" + + def run_purged_fold( + self, + strategy_profile: str, + params: Mapping[str, Any], + *, + fold: PurgedWalkForwardFold, + purge_days: int, + embargo_days: int, + cost_model: PromotionCostModel, + ) -> BacktestResult: ... + + def run_locked_oos( + self, + strategy_profile: str, + params: Mapping[str, Any], + *, + start_date: date, + end_date: date, + cost_model: PromotionCostModel, + ) -> BacktestResult: ... + + class BacktestOrchestrator: """Orchestrates backtest runs across strategies and markets. @@ -83,35 +268,44 @@ def persist_result( param_set_id: str = "", param_version: int | None = None, ) -> BacktestResult: - enriched = BacktestResult( + return self._persist_result( + result, + strategy_profile=strategy_profile, + domain=domain, + params=params, + param_set_id=param_set_id, + param_version=param_version, + validation_identity=None, + ) + + def _persist_result( + self, + result: BacktestResult, + *, + strategy_profile: str, + domain: str, + params: Mapping[str, Any], + param_set_id: str, + param_version: int | None, + validation_identity: BacktestValidationIdentity | None, + ) -> BacktestResult: + enriched = replace( + result, strategy_profile=strategy_profile, domain=domain, param_set_id=param_set_id or result.param_set_id or _run_id(), params=dict(params), - param_version=max(int((result.param_version if param_version is None else param_version) or 1), 1), - sharpe_ratio=result.sharpe_ratio, - calmar_ratio=result.calmar_ratio, - sortino_ratio=result.sortino_ratio, - max_drawdown=result.max_drawdown, - cagr=result.cagr, - volatility=result.volatility, - win_rate=result.win_rate, - total_return=result.total_return, - start_date=result.start_date, - end_date=result.end_date, - observation_count=result.observation_count, - benchmark_symbol=result.benchmark_symbol, - benchmark_cagr=result.benchmark_cagr, - benchmark_max_drawdown=result.benchmark_max_drawdown, - excess_cagr=result.excess_cagr, - oos_sharpe=result.oos_sharpe, - oos_calmar=result.oos_calmar, - oos_max_drawdown=result.oos_max_drawdown, - walk_forward_stability=result.walk_forward_stability, + param_version=max( + int( + (result.param_version if param_version is None else param_version) + or 1 + ), + 1, + ), run_id=result.run_id or _run_id(), - run_duration_seconds=result.run_duration_seconds, source_script=result.source_script or "backtest_orchestrator", computed_at=result.computed_at or _now_iso(), + validation_identity=validation_identity, ) self._store.save_backtest_result(enriched) return enriched @@ -146,9 +340,13 @@ def run( """ runner = self._runners.get(domain) if runner is None: - raise ValueError(f"No BacktestRunner registered for domain={domain!r}. Available: {sorted(self._runners)}") + raise ValueError( + f"No BacktestRunner registered for domain={domain!r}. Available: {sorted(self._runners)}" + ) - result = runner.run(strategy_profile, params, start_date=start_date, end_date=end_date) + result = runner.run( + strategy_profile, params, start_date=start_date, end_date=end_date + ) if result.start_date is None or result.end_date is None: result = replace( result, @@ -164,10 +362,162 @@ def run( param_version=param_version, ) - def run_latest(self, strategy_profile: str, *, domain: str) -> BacktestResult | None: + def run_latest( + self, strategy_profile: str, *, domain: str + ) -> BacktestResult | None: """Load the latest persisted backtest result for a strategy.""" return self._store.load_latest_backtest(domain, strategy_profile) + def run_promotion( + self, + strategy_profile: str, + *, + domain: str, + params: Mapping[str, Any], + folds: Sequence[PurgedWalkForwardFold], + locked_oos_start: date, + locked_oos_end: date, + purge_days: int, + embargo_days: int, + source_revision: str, + cost_model: PromotionCostModel, + param_set_id: str = "", + param_version: int = 1, + ) -> PromotionBacktestRun: + """Run the explicit promotion-grade Purged Walk-Forward protocol. + + Ordinary ``run`` and ``walk_forward`` calls remain non-promotion. A + runner opts in only by implementing both explicit promotion methods; + generic ``**kwargs`` support is not treated as a capability. + """ + validated_folds = _validate_promotion_plan( + folds, + locked_oos_start=locked_oos_start, + locked_oos_end=locked_oos_end, + purge_days=purge_days, + embargo_days=embargo_days, + source_revision=source_revision, + cost_model=cost_model, + ) + runner = self._runners.get(domain) + if runner is None: + raise ValueError( + f"No BacktestRunner registered for domain={domain!r}. Available: {sorted(self._runners)}" + ) + if not isinstance(runner, PromotionBacktestRunner): + raise TypeError( + "promotion-grade execution requires explicit run_purged_fold and run_locked_oos runner methods" + ) + + raw_fold_results: list[BacktestResult] = [] + for fold in validated_folds: + raw_fold_results.append( + _validate_promotion_result( + runner.run_purged_fold( + strategy_profile, + params, + fold=fold, + purge_days=purge_days, + embargo_days=embargo_days, + cost_model=cost_model, + ), + start_date=fold.test_start, + end_date=fold.test_end, + ) + ) + raw_locked_oos_result = _validate_promotion_result( + runner.run_locked_oos( + strategy_profile, + params, + start_date=locked_oos_start, + end_date=locked_oos_end, + cost_model=cost_model, + ), + start_date=locked_oos_start, + end_date=locked_oos_end, + ) + + base_id = param_set_id or _run_id() + cost_inputs = { + "commission_bps": float(cost_model.commission_bps), + "slippage_bps": float(cost_model.slippage_bps), + "market_impact_bps": float(cost_model.market_impact_bps), + } + fold_results: list[BacktestResult] = [] + for index, (fold, raw_result) in enumerate( + zip(validated_folds, raw_fold_results) + ): + identity = BacktestValidationIdentity( + protocol=_PROMOTION_PROTOCOL, + fold_id=f"{base_id}_wf{index}", + fold_role="test", + train_start=fold.train_start, + train_end=fold.train_end, + test_start=fold.test_start, + test_end=fold.test_end, + locked_oos_start=locked_oos_start, + locked_oos_end=locked_oos_end, + purge_days=purge_days, + embargo_days=embargo_days, + ) + fold_results.append( + self._persist_result( + replace( + raw_result, + source_revision=source_revision, + cost_model=cost_model.model_id, + cost_inputs=cost_inputs, + ), + strategy_profile=strategy_profile, + domain=domain, + params=params, + param_set_id=identity.fold_id, + param_version=param_version, + validation_identity=identity, + ) + ) + + locked_identity = BacktestValidationIdentity( + protocol=_PROMOTION_PROTOCOL, + fold_id=f"{base_id}_locked_oos", + fold_role="locked_oos", + train_start=None, + train_end=None, + test_start=locked_oos_start, + test_end=locked_oos_end, + locked_oos_start=locked_oos_start, + locked_oos_end=locked_oos_end, + purge_days=purge_days, + embargo_days=embargo_days, + ) + locked_oos_result = self._persist_result( + replace( + raw_locked_oos_result, + source_revision=source_revision, + cost_model=cost_model.model_id, + cost_inputs=cost_inputs, + ), + strategy_profile=strategy_profile, + domain=domain, + params=params, + param_set_id=locked_identity.fold_id, + param_version=param_version, + validation_identity=locked_identity, + ) + return PromotionBacktestRun( + strategy_profile=strategy_profile, + domain=domain, + fold_results=tuple(fold_results), + locked_oos_result=locked_oos_result, + folds=validated_folds, + locked_oos_start=locked_oos_start, + locked_oos_end=locked_oos_end, + purge_days=purge_days, + embargo_days=embargo_days, + source_revision=source_revision, + cost_model=cost_model, + ) + def walk_forward( self, strategy_profile: str, @@ -195,7 +545,9 @@ def walk_forward( ValueError: If windows is empty or no runner is registered. """ if not windows: - raise ValueError("windows must contain at least one (start_date, end_date) pair") + raise ValueError( + "windows must contain at least one (start_date, end_date) pair" + ) base_id = param_set_id or _run_id() results: list[BacktestResult] = [] @@ -245,7 +597,9 @@ def sensitivity( ValueError: If param_ranges is empty or no runner is registered. """ if not param_ranges: - raise ValueError("param_ranges must contain at least one parameter dimension") + raise ValueError( + "param_ranges must contain at least one parameter dimension" + ) keys = sorted(param_ranges.keys()) value_lists = [list(param_ranges[k]) for k in keys] @@ -255,7 +609,10 @@ def sensitivity( combos: list[dict[str, Any]] = [] for idx, combo in enumerate(itertools.product(*value_lists)): - if total > max_combinations and idx % max(1, total // max_combinations) != 0: + if ( + total > max_combinations + and idx % max(1, total // max_combinations) != 0 + ): continue merged = dict(base_params) merged.update(dict(zip(keys, combo))) diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index dfb7921f..d7487328 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -145,7 +145,12 @@ def from_score(cls, score: float) -> "DriftStatus": @property def severity_order(self) -> int: - _order = {DriftStatus.HEALTHY: 0, DriftStatus.WATCH: 1, DriftStatus.REVIEW: 2, DriftStatus.CRITICAL: 3} + _order = { + DriftStatus.HEALTHY: 0, + DriftStatus.WATCH: 1, + DriftStatus.REVIEW: 2, + DriftStatus.CRITICAL: 3, + } return _order[self] @@ -200,7 +205,9 @@ def to_dict(self) -> dict[str, object]: "drift_score": self.drift_score, "status": self.status.value, "dimensions": {k: v.to_dict() for k, v in self.dimensions.items()}, - "previous_status": self.previous_status.value if self.previous_status else None, + "previous_status": self.previous_status.value + if self.previous_status + else None, "baseline_param_set_id": self.baseline_param_set_id, "baseline_available": self.baseline_available, "baseline_param_version": self.baseline_param_version, @@ -218,6 +225,74 @@ def breached_dimensions(self) -> tuple[DriftDimension, ...]: # ── Backtest & Optimization ───────────────────────────────────────── +@dataclass(frozen=True) +class PurgedWalkForwardFold: + """Explicit train/test boundaries for one promotion-grade fold.""" + + train_start: date + train_end: date + test_start: date + test_end: date + + def to_dict(self) -> dict[str, str]: + return { + "train_start": self.train_start.isoformat(), + "train_end": self.train_end.isoformat(), + "test_start": self.test_start.isoformat(), + "test_end": self.test_end.isoformat(), + } + + +@dataclass(frozen=True) +class PromotionCostModel: + """Finite cost inputs required by promotion-grade orchestration.""" + + model_id: str + commission_bps: float + slippage_bps: float + market_impact_bps: float = 0.0 + + def to_dict(self) -> dict[str, str | float]: + return { + "model_id": self.model_id, + "commission_bps": self.commission_bps, + "slippage_bps": self.slippage_bps, + "market_impact_bps": self.market_impact_bps, + } + + +@dataclass(frozen=True) +class BacktestValidationIdentity: + """Orchestrator-computed timing identity; never a caller promotion flag.""" + + protocol: str + fold_id: str + fold_role: str + train_start: date | None + train_end: date | None + test_start: date + test_end: date + locked_oos_start: date + locked_oos_end: date + purge_days: int + embargo_days: int + + def to_dict(self) -> dict[str, object]: + return { + "protocol": self.protocol, + "fold_id": self.fold_id, + "fold_role": self.fold_role, + "train_start": self.train_start.isoformat() if self.train_start else None, + "train_end": self.train_end.isoformat() if self.train_end else None, + "test_start": self.test_start.isoformat(), + "test_end": self.test_end.isoformat(), + "locked_oos_start": self.locked_oos_start.isoformat(), + "locked_oos_end": self.locked_oos_end.isoformat(), + "purge_days": self.purge_days, + "embargo_days": self.embargo_days, + } + + @dataclass(frozen=True) class BacktestResult: """Standardized result from a single backtest run.""" @@ -263,6 +338,10 @@ class BacktestResult: source_revision: str = "" cost_model: str = "" + # Appended to preserve the positional order of every legacy field above. + validation_identity: BacktestValidationIdentity | None = None + cost_inputs: Mapping[str, float] = field(default_factory=dict) + def to_dict(self) -> dict[str, object]: return { "strategy_profile": self.strategy_profile, @@ -295,6 +374,42 @@ def to_dict(self) -> dict[str, object]: "computed_at": self.computed_at, "source_revision": self.source_revision, "cost_model": self.cost_model, + "validation_identity": self.validation_identity.to_dict() + if self.validation_identity + else None, + "cost_inputs": dict(self.cost_inputs), + } + + +@dataclass(frozen=True) +class PromotionBacktestRun: + """Validated output produced only by the strict promotion orchestration path.""" + + strategy_profile: str + domain: str + fold_results: tuple[BacktestResult, ...] + locked_oos_result: BacktestResult + folds: tuple[PurgedWalkForwardFold, ...] + locked_oos_start: date + locked_oos_end: date + purge_days: int + embargo_days: int + source_revision: str + cost_model: PromotionCostModel + + def to_dict(self) -> dict[str, object]: + return { + "strategy_profile": self.strategy_profile, + "domain": self.domain, + "folds": [fold.to_dict() for fold in self.folds], + "fold_results": [result.to_dict() for result in self.fold_results], + "locked_oos_result": self.locked_oos_result.to_dict(), + "locked_oos_start": self.locked_oos_start.isoformat(), + "locked_oos_end": self.locked_oos_end.isoformat(), + "purge_days": self.purge_days, + "embargo_days": self.embargo_days, + "source_revision": self.source_revision, + "cost_model": self.cost_model.to_dict(), } @@ -391,9 +506,13 @@ def to_dict(self) -> dict[str, object]: "strategy_profile": self.strategy_profile, "domain": self.domain, "current_params": dict(self.current_params), - "current_metrics": self.current_metrics.to_dict() if self.current_metrics else None, + "current_metrics": self.current_metrics.to_dict() + if self.current_metrics + else None, "proposed_params": dict(self.proposed_params), - "proposed_metrics": self.proposed_metrics.to_dict() if self.proposed_metrics else None, + "proposed_metrics": self.proposed_metrics.to_dict() + if self.proposed_metrics + else None, "improvement_score": self.improvement_score, "confidence": self.confidence, "winning_dimensions": list(self.winning_dimensions), diff --git a/tests/test_backtest_orchestrator.py b/tests/test_backtest_orchestrator.py index 4dcd34bd..9fbbbf06 100644 --- a/tests/test_backtest_orchestrator.py +++ b/tests/test_backtest_orchestrator.py @@ -2,13 +2,18 @@ from __future__ import annotations +from dataclasses import replace from datetime import date +import json +import math import tempfile import unittest from pathlib import Path from typing import Any, Mapping -from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator +from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import ( + BacktestOrchestrator, +) from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore @@ -50,8 +55,93 @@ def run( ) -class BacktestOrchestratorTests(unittest.TestCase): +class _PromotionRecordingRunner(_RecordingRunner): + """Synthetic runner with the explicit promotion-only capability.""" + + def __init__(self, **result_overrides: Any) -> None: + super().__init__() + self.result_overrides = result_overrides + + def _promotion_result( + self, + strategy_profile: str, + params: Mapping[str, Any], + *, + start_date: date, + end_date: date, + ) -> BacktestResult: + values: dict[str, Any] = { + "strategy_profile": strategy_profile, + "domain": "us_equity", + "param_set_id": "runner_result", + "params": dict(params), + "sharpe_ratio": 1.2, + "calmar_ratio": 1.1, + "sortino_ratio": 1.4, + "max_drawdown": -0.08, + "cagr": 0.12, + "volatility": 0.15, + "win_rate": 0.55, + "total_return": 0.18, + "start_date": start_date, + "end_date": end_date, + "observation_count": 252, + } + values.update(self.result_overrides) + return BacktestResult(**values) + + def run_purged_fold( + self, + strategy_profile: str, + params: Mapping[str, Any], + *, + fold: Any, + purge_days: int, + embargo_days: int, + cost_model: Any, + ) -> BacktestResult: + self.calls.append( + { + "kind": "fold", + "fold": fold, + "purge_days": purge_days, + "embargo_days": embargo_days, + "cost_model": cost_model, + } + ) + return self._promotion_result( + strategy_profile, + params, + start_date=fold.test_start, + end_date=fold.test_end, + ) + def run_locked_oos( + self, + strategy_profile: str, + params: Mapping[str, Any], + *, + start_date: date, + end_date: date, + cost_model: Any, + ) -> BacktestResult: + self.calls.append( + { + "kind": "locked_oos", + "start_date": start_date, + "end_date": end_date, + "cost_model": cost_model, + } + ) + return self._promotion_result( + strategy_profile, + params, + start_date=start_date, + end_date=end_date, + ) + + +class BacktestOrchestratorTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.store = PerformanceStore(local_root=Path(self.tmp.name)) @@ -62,6 +152,74 @@ def setUp(self) -> None: def tearDown(self) -> None: self.tmp.cleanup() + @staticmethod + def _promotion_folds() -> list[Any]: + from quant_platform_kit.strategy_lifecycle.contracts import ( + PurgedWalkForwardFold, + ) + + return [ + PurgedWalkForwardFold( + date(2015, 1, 1), + date(2015, 12, 31), + date(2016, 1, 3), + date(2016, 6, 30), + ), + PurgedWalkForwardFold( + date(2016, 7, 3), + date(2017, 6, 30), + date(2017, 7, 3), + date(2017, 12, 31), + ), + PurgedWalkForwardFold( + date(2018, 1, 3), + date(2018, 12, 31), + date(2019, 1, 3), + date(2019, 6, 30), + ), + ] + + @staticmethod + def _cost_model(**overrides: Any) -> Any: + from quant_platform_kit.strategy_lifecycle.contracts import PromotionCostModel + + values = { + "model_id": "retail_us_equity_v1", + "commission_bps": 1.0, + "slippage_bps": 2.0, + "market_impact_bps": 0.5, + } + values.update(overrides) + return PromotionCostModel(**values) + + def _run_promotion( + self, + *, + runner: _PromotionRecordingRunner | None = None, + folds: list[Any] | None = None, + locked_oos_start: date = date(2019, 7, 3), + locked_oos_end: date = date(2020, 7, 3), + purge_days: int = 1, + embargo_days: int = 1, + source_revision: str = "a" * 40, + cost_model: Any = None, + ) -> Any: + promotion_runner = runner or _PromotionRecordingRunner() + self.orchestrator.register_runner("us_equity", promotion_runner) + return self.orchestrator.run_promotion( + "test_strat", + domain="us_equity", + params={"lookback": 30}, + folds=self._promotion_folds() if folds is None else folds, + locked_oos_start=locked_oos_start, + locked_oos_end=locked_oos_end, + purge_days=purge_days, + embargo_days=embargo_days, + source_revision=source_revision, + cost_model=cost_model or self._cost_model(), + param_set_id="promotion_test", + ) + def test_run_enriches_and_persists(self) -> None: result = self.orchestrator.run( "test_strat", @@ -125,6 +283,35 @@ def test_persist_result_clamps_explicit_zero_param_version(self) -> None: self.assertEqual(persisted.param_version, 1) + def test_persist_result_preserves_source_revision_and_cost_model(self) -> None: + result = BacktestResult( + strategy_profile="test_strat", + domain="us_equity", + param_set_id="candidate", + params={"lookback": 30}, + source_revision="b" * 40, + cost_model="retail_us_equity_v1", + ) + + persisted = self.orchestrator.persist_result( + result, + strategy_profile="test_strat", + domain="us_equity", + params={"lookback": 30}, + ) + + self.assertEqual(persisted.source_revision, "b" * 40) + self.assertEqual(persisted.cost_model, "retail_us_equity_v1") + + def test_backtest_result_keeps_legacy_positional_order(self) -> None: + result = BacktestResult( + "strat", "domain", "params-v1", {"lookback": 30}, 7, 1.25, 0.9 + ) + + self.assertEqual(result.param_version, 7) + self.assertEqual(result.sharpe_ratio, 1.25) + self.assertEqual(result.calmar_ratio, 0.9) + def test_run_raises_without_runner(self) -> None: with self.assertRaises(ValueError): self.orchestrator.run("test_strat", domain="cn_equity", params={}) @@ -162,6 +349,242 @@ def test_walk_forward_empty_windows_raises(self) -> None: windows=[], ) + def test_ordinary_walk_forward_accepts_raw_windows_but_is_non_promotion( + self, + ) -> None: + results = self.orchestrator.walk_forward( + "test_strat", + domain="us_equity", + params={}, + windows=[ + (date(2021, 1, 1), date(2021, 12, 31)), + (date(2021, 6, 1), date(2021, 7, 1)), + (None, None), + ], + ) + + self.assertEqual(len(results), 3) + self.assertTrue( + all( + getattr(result, "validation_identity", None) is None + for result in results + ) + ) + self.assertTrue( + all("promotion_ready" not in result.to_dict() for result in results) + ) + + def test_promotion_run_requires_explicit_runner_capability(self) -> None: + with self.assertRaises(TypeError): + self.orchestrator.run_promotion( + "test_strat", + domain="us_equity", + params={}, + folds=self._promotion_folds(), + locked_oos_start=date(2019, 7, 3), + locked_oos_end=date(2020, 7, 3), + purge_days=1, + embargo_days=1, + source_revision="a" * 40, + cost_model=self._cost_model(), + ) + + def test_promotion_run_enforces_and_persists_purged_wfa_identity(self) -> None: + run = self._run_promotion() + + self.assertEqual(len(run.fold_results), 3) + self.assertEqual(run.locked_oos_result.start_date, date(2019, 7, 3)) + self.assertEqual(run.locked_oos_result.end_date, date(2020, 7, 3)) + self.assertEqual(run.source_revision, "a" * 40) + self.assertEqual(run.cost_model.model_id, "retail_us_equity_v1") + results = (*run.fold_results, run.locked_oos_result) + self.assertTrue(all(result.source_revision == "a" * 40 for result in results)) + self.assertTrue( + all(result.cost_model == "retail_us_equity_v1" for result in results) + ) + self.assertTrue( + all(result.validation_identity is not None for result in results) + ) + self.assertEqual( + [result.validation_identity.fold_role for result in run.fold_results], + ["test"] * 3, + ) + self.assertEqual( + run.locked_oos_result.validation_identity.fold_role, "locked_oos" + ) + self.assertTrue( + all(result.validation_identity.purge_days == 1 for result in results) + ) + self.assertTrue( + all(result.validation_identity.embargo_days == 1 for result in results) + ) + self.assertTrue( + all(result.cost_inputs["commission_bps"] == 1.0 for result in results) + ) + + payloads = [ + json.loads(path.read_text()) for path in Path(self.tmp.name).rglob("*.json") + ] + self.assertEqual(len(payloads), 4) + self.assertTrue( + all(payload["source_revision"] == "a" * 40 for payload in payloads) + ) + self.assertTrue( + all(payload["cost_model"] == "retail_us_equity_v1" for payload in payloads) + ) + self.assertTrue( + all( + payload["validation_identity"]["purge_days"] == 1 + for payload in payloads + ) + ) + self.assertTrue( + all( + payload["validation_identity"]["embargo_days"] == 1 + for payload in payloads + ) + ) + + def test_promotion_run_rejects_untyped_raw_windows(self) -> None: + with self.assertRaises(TypeError): + self._run_promotion(folds=[(date(2015, 1, 1), date(2016, 1, 1))]) + + def test_promotion_run_rejects_fewer_than_three_folds(self) -> None: + with self.assertRaises(ValueError): + self._run_promotion(folds=self._promotion_folds()[:2]) + + def test_promotion_run_rejects_zero_purge_or_embargo(self) -> None: + with self.assertRaises(ValueError): + self._run_promotion(purge_days=0) + with self.assertRaises(ValueError): + self._run_promotion(embargo_days=0) + + def test_promotion_run_requires_explicit_embargo(self) -> None: + runner = _PromotionRecordingRunner() + self.orchestrator.register_runner("us_equity", runner) + with self.assertRaises(TypeError): + self.orchestrator.run_promotion( + "test_strat", + domain="us_equity", + params={}, + folds=self._promotion_folds(), + locked_oos_start=date(2019, 7, 3), + locked_oos_end=date(2020, 7, 3), + purge_days=1, + source_revision="a" * 40, + cost_model=self._cost_model(), + ) + + def test_promotion_run_rejects_invalid_or_overlapping_folds(self) -> None: + fold_type = type(self._promotion_folds()[0]) + invalid_sets = [ + [ + fold_type( + date(2015, 1, 2), + date(2015, 1, 1), + date(2015, 2, 1), + date(2015, 3, 1), + ), + *self._promotion_folds()[1:], + ], + [ + fold_type( + date(2015, 1, 1), + date(2015, 12, 31), + date(2015, 12, 31), + date(2016, 6, 30), + ), + *self._promotion_folds()[1:], + ], + [ + self._promotion_folds()[1], + self._promotion_folds()[0], + self._promotion_folds()[2], + ], + [ + self._promotion_folds()[0], + fold_type( + date(2016, 6, 29), + date(2017, 6, 30), + date(2017, 7, 3), + date(2017, 12, 31), + ), + self._promotion_folds()[2], + ], + ] + for folds in invalid_sets: + with self.subTest(folds=folds), self.assertRaises(ValueError): + self._run_promotion(folds=folds) + + def test_promotion_run_rejects_locked_oos_overlap_or_short_calendar_span( + self, + ) -> None: + with self.assertRaises(ValueError): + self._run_promotion(locked_oos_start=date(2019, 6, 30)) + with self.assertRaises(ValueError): + self._run_promotion(locked_oos_end=date(2020, 7, 2)) + + def test_promotion_run_uses_calendar_month_validation_for_leap_day(self) -> None: + run = self._run_promotion( + locked_oos_start=date(2020, 2, 29), + locked_oos_end=date(2021, 2, 28), + ) + + self.assertEqual(run.locked_oos_result.end_date, date(2021, 2, 28)) + + def test_promotion_run_rejects_undated_or_non_finite_results(self) -> None: + with self.assertRaises(ValueError): + self._run_promotion(runner=_PromotionRecordingRunner(start_date=None)) + with self.assertRaises(ValueError): + self._run_promotion(runner=_PromotionRecordingRunner(sharpe_ratio=math.nan)) + with self.assertRaises(ValueError): + self._run_promotion(runner=_PromotionRecordingRunner(cagr=math.inf)) + + def test_promotion_run_rejects_non_finite_cost_inputs(self) -> None: + with self.assertRaises(ValueError): + self._run_promotion(cost_model=self._cost_model(slippage_bps=math.nan)) + with self.assertRaises(ValueError): + self._run_promotion(cost_model=self._cost_model(commission_bps=math.inf)) + + def test_promotion_run_rejects_invalid_source_revision(self) -> None: + with self.assertRaises(ValueError): + self._run_promotion(source_revision="caller-label") + + def test_public_persist_cannot_promote_a_caller_labeled_result(self) -> None: + from quant_platform_kit.strategy_lifecycle.contracts import ( + BacktestValidationIdentity, + ) + + fold = self._promotion_folds()[0] + identity = BacktestValidationIdentity( + protocol="purged_walk_forward.v1", + fold_id="caller", + fold_role="locked_oos", + train_start=None, + train_end=None, + test_start=date(2019, 7, 3), + test_end=date(2020, 7, 3), + locked_oos_start=date(2019, 7, 3), + locked_oos_end=date(2020, 7, 3), + purge_days=1, + embargo_days=1, + ) + caller_labeled = replace( + self.runner.run( + "test_strat", {}, start_date=fold.test_start, end_date=fold.test_end + ), + validation_identity=identity, + ) + + persisted = self.orchestrator.persist_result( + caller_labeled, + strategy_profile="test_strat", + domain="us_equity", + params={}, + ) + + self.assertIsNone(persisted.validation_identity) + def test_sensitivity_runs_param_grid(self) -> None: report = self.orchestrator.sensitivity( "test_strat",