diff --git a/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py b/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py index efedb95..4c695c0 100644 --- a/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py +++ b/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py @@ -6,6 +6,7 @@ from __future__ import annotations import itertools +import inspect import uuid from dataclasses import replace from datetime import date, datetime, timezone @@ -14,6 +15,9 @@ from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult, SensitivityReport from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore +EXECUTION_TIMINGS = ("next_open", "next_close") +_UNSET_TIMING = object() + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -37,6 +41,8 @@ def run( params: Mapping[str, Any], start_date: date | None = None, end_date: date | None = None, + execution_timing: str | None = None, + persist: bool = True, ) -> BacktestResult: """Execute a backtest for the given strategy with the given parameters. @@ -82,6 +88,8 @@ def persist_result( params: Mapping[str, Any], param_set_id: str = "", param_version: int | None = None, + save: bool = True, + execution_timing: str | None = None, ) -> BacktestResult: enriched = BacktestResult( strategy_profile=strategy_profile, @@ -89,6 +97,7 @@ def persist_result( 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), + execution_timing=execution_timing if execution_timing is not None else result.execution_timing, sharpe_ratio=result.sharpe_ratio, calmar_ratio=result.calmar_ratio, sortino_ratio=result.sortino_ratio, @@ -113,9 +122,33 @@ def persist_result( source_script=result.source_script or "backtest_orchestrator", computed_at=result.computed_at or _now_iso(), ) - self._store.save_backtest_result(enriched) + if save: + self._store.save_backtest_result(enriched) return enriched + @staticmethod + def _run_runner( + runner: BacktestRunner, + strategy_profile: str, + params: Mapping[str, Any], + *, + start_date: date | None, + end_date: date | None, + execution_timing: str | None, + persist: bool, + ) -> BacktestResult: + kwargs: dict[str, Any] = {"start_date": start_date, "end_date": end_date} + if execution_timing is not None: + parameters = inspect.signature(runner.run).parameters + if "execution_timing" not in parameters: + raise TypeError("runner must accept execution_timing for explicit timing semantics") + kwargs["execution_timing"] = execution_timing + if not persist: + if "persist" not in inspect.signature(runner.run).parameters: + raise TypeError("runner must accept persist=False for ephemeral execution") + kwargs["persist"] = False + return runner.run(strategy_profile, params, **kwargs) + def run( self, strategy_profile: str, @@ -126,6 +159,8 @@ def run( param_version: int = 1, start_date: date | None = None, end_date: date | None = None, + execution_timing: str | None = None, + persist: bool = True, ) -> BacktestResult: """Run a backtest for a strategy. @@ -144,11 +179,21 @@ def run( Raises: ValueError: If no runner is registered for the domain. """ + if execution_timing is not None and execution_timing not in EXECUTION_TIMINGS: + raise ValueError(f"unsupported execution_timing={execution_timing!r}; expected one of {EXECUTION_TIMINGS}") runner = self._runners.get(domain) if runner is None: 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 = self._run_runner( + runner, + strategy_profile, + params, + start_date=start_date, + end_date=end_date, + execution_timing=execution_timing, + persist=persist, + ) if result.start_date is None or result.end_date is None: result = replace( result, @@ -162,11 +207,21 @@ def run( params=params, param_set_id=param_set_id, param_version=param_version, + save=persist, + execution_timing=execution_timing, ) - def run_latest(self, strategy_profile: str, *, domain: str) -> BacktestResult | None: + def run_latest( + self, + strategy_profile: str, + *, + domain: str, + execution_timing: str | None | object = _UNSET_TIMING, + ) -> BacktestResult | None: """Load the latest persisted backtest result for a strategy.""" - return self._store.load_latest_backtest(domain, strategy_profile) + if execution_timing is _UNSET_TIMING: + return self._store.load_latest_backtest(domain, strategy_profile) + return self._store.load_latest_backtest(domain, strategy_profile, execution_timing=execution_timing) def walk_forward( self, @@ -177,6 +232,8 @@ def walk_forward( windows: Sequence[tuple[date | None, date | None]], param_set_id: str = "", param_version: int = 1, + execution_timing: str | None = None, + persist: bool = True, ) -> list[BacktestResult]: """Run backtests across multiple time windows. @@ -209,6 +266,8 @@ def walk_forward( param_version=param_version, start_date=start_date, end_date=end_date, + execution_timing=execution_timing, + persist=persist, ) ) return results diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index dfb7921..f661c26 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/src/quant_platform_kit/strategy_lifecycle/contracts.py @@ -227,6 +227,7 @@ class BacktestResult: param_set_id: str params: Mapping[str, Any] param_version: int = 1 + execution_timing: str | None = None # Core metrics sharpe_ratio: float | None = None @@ -270,6 +271,7 @@ def to_dict(self) -> dict[str, object]: "param_set_id": self.param_set_id, "params": dict(self.params), "param_version": self.param_version, + "execution_timing": self.execution_timing, "sharpe_ratio": self.sharpe_ratio, "calmar_ratio": self.calmar_ratio, "sortino_ratio": self.sortino_ratio, diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 4fbe735..b7847d2 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -34,6 +34,7 @@ SCHEMA_VERSION = "strategy_lifecycle.v1" DEFAULT_BUCKET_ENV = "LIFECYCLE_PERFORMANCE_BUCKET" DEFAULT_LOCAL_ROOT = Path(tempfile.gettempdir()) / "quant_platform_lifecycle" +_UNSET_TIMING = object() def _now_iso() -> str: @@ -265,7 +266,13 @@ def save_backtest_result(self, result: BacktestResult) -> None: {**result.to_dict(), "schema_version": SCHEMA_VERSION}, ) - def load_latest_backtest(self, domain: str, strategy_profile: str) -> BacktestResult | None: + def load_latest_backtest( + self, + domain: str, + strategy_profile: str, + *, + execution_timing: str | None | object = _UNSET_TIMING, + ) -> BacktestResult | None: prefix = f"backtest/{_clean_key(domain)}/{_clean_key(strategy_profile)}/" keys = list(dict.fromkeys([*self._list_cloud_keys(prefix), *self._list_local_json_keys(prefix)])) if not keys: @@ -279,6 +286,10 @@ def load_latest_backtest(self, domain: str, strategy_profile: str) -> BacktestRe candidates.append((_backtest_sort_key(result, key), result)) if not candidates: return None + if execution_timing is not _UNSET_TIMING: + candidates = [item for item in candidates if item[1].execution_timing == execution_timing] + if not candidates: + return None baseline_candidates = [item for item in candidates if _is_baseline_backtest(item[1])] selected = baseline_candidates or candidates selected.sort(key=lambda item: item[0]) @@ -563,6 +574,7 @@ def _backtest_from_dict(data: Mapping[str, Any]) -> BacktestResult | None: param_set_id=str(data.get("param_set_id", "")), params=dict(data.get("params", {})), param_version=int(data.get("param_version", 1)), + execution_timing=(str(data["execution_timing"]) if data.get("execution_timing") else None), sharpe_ratio=float(data["sharpe_ratio"]) if data.get("sharpe_ratio") is not None else None, calmar_ratio=float(data["calmar_ratio"]) if data.get("calmar_ratio") is not None else None, sortino_ratio=float(data["sortino_ratio"]) if data.get("sortino_ratio") is not None else None, diff --git a/tests/test_backtest_orchestrator.py b/tests/test_backtest_orchestrator.py index 4dcd34b..7976bcc 100644 --- a/tests/test_backtest_orchestrator.py +++ b/tests/test_backtest_orchestrator.py @@ -50,6 +50,21 @@ def run( ) +class _TimingRunner(_RecordingRunner): + def run( + self, + strategy_profile: str, + params: Mapping[str, Any], + start_date: date | None = None, + end_date: date | None = None, + *, + execution_timing: str, + persist: bool = True, + ) -> BacktestResult: + self.calls.append({"execution_timing": execution_timing, "persist": persist}) + return super().run(strategy_profile, params, start_date, end_date) + + class BacktestOrchestratorTests(unittest.TestCase): def setUp(self) -> None: @@ -78,6 +93,64 @@ def test_run_enriches_and_persists(self) -> None: self.assertTrue(result.computed_at) self.assertEqual(result.source_script, "backtest_orchestrator") + def test_run_forwards_timing_and_can_skip_persistence(self) -> None: + runner = _TimingRunner() + orchestrator = BacktestOrchestrator(store=self.store) + orchestrator.register_runner("us_equity", runner) + + result = orchestrator.run( + "soxl_soxx_trend_income", + domain="us_equity", + params={"lookback": 20}, + execution_timing="next_open", + persist=False, + ) + + self.assertEqual(runner.calls[0]["execution_timing"], "next_open") + self.assertFalse(runner.calls[0]["persist"]) + self.assertIsNone(orchestrator.run_latest(result.strategy_profile, domain="us_equity")) + + def test_persisted_timing_is_part_of_result_identity(self) -> None: + runner = _TimingRunner() + orchestrator = BacktestOrchestrator(store=self.store) + orchestrator.register_runner("us_equity", runner) + + orchestrator.run("test_strat", domain="us_equity", params={}, execution_timing="next_open") + orchestrator.run("test_strat", domain="us_equity", params={}, execution_timing="next_close") + + self.assertEqual( + orchestrator.run_latest("test_strat", domain="us_equity", execution_timing="next_open").execution_timing, + "next_open", + ) + self.assertEqual( + orchestrator.run_latest("test_strat", domain="us_equity", execution_timing="next_close").execution_timing, + "next_close", + ) + self.assertEqual(orchestrator.run_latest("test_strat", domain="us_equity").execution_timing, "next_close") + self.assertIsNone(orchestrator.run_latest("test_strat", domain="us_equity", execution_timing=None)) + + def test_explicit_timing_rejects_kwargs_only_runner(self) -> None: + class KeywordRunner(_RecordingRunner): + def run(self, strategy_profile, params, **kwargs): + return super().run(strategy_profile, params) + + orchestrator = BacktestOrchestrator(store=self.store) + orchestrator.register_runner("us_equity", KeywordRunner()) + with self.assertRaisesRegex(TypeError, "execution_timing"): + orchestrator.run( + "test_strat", domain="us_equity", params={}, execution_timing="next_close", persist=False + ) + + def test_ephemeral_mode_requires_runner_persist_capability(self) -> None: + with self.assertRaisesRegex(TypeError, "persist"): + self.orchestrator.run("test_strat", domain="us_equity", params={}, persist=False) + + def test_run_rejects_unknown_timing(self) -> None: + with self.assertRaises(ValueError): + self.orchestrator.run( + "test_strat", domain="us_equity", params={}, execution_timing="same_close", persist=False + ) + def test_persist_result_preserves_existing_param_version_by_default(self) -> None: result = BacktestResult( strategy_profile="test_strat",