feat: add R1 timing semantics to BacktestOrchestrator

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,13 +88,16 @@ 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,
             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),
+            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

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,

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,

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",