feat: migrate backtest result timing identity

src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py

@@ -89,6 +89,9 @@ 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=result.execution_timing,
+            result_identity_version=result.result_identity_version,
+            persist_mode=result.persist_mode,
             sharpe_ratio=result.sharpe_ratio,
             calmar_ratio=result.calmar_ratio,
             sortino_ratio=result.sortino_ratio,

src/quant_platform_kit/strategy_lifecycle/contracts.py

@@ -262,6 +262,9 @@ class BacktestResult:
     computed_at: str = ""
     source_revision: str = ""
     cost_model: str = ""
+    execution_timing: str | None = field(default=None, kw_only=True)
+    result_identity_version: int = field(default=1, kw_only=True)
+    persist_mode: str = field(default="durable", kw_only=True)
 
     def to_dict(self) -> dict[str, object]:
         return {
@@ -295,6 +298,9 @@ def to_dict(self) -> dict[str, object]:
             "computed_at": self.computed_at,
             "source_revision": self.source_revision,
             "cost_model": self.cost_model,
+            "execution_timing": self.execution_timing,
+            "result_identity_version": self.result_identity_version,
+            "persist_mode": self.persist_mode,
         }
 
 

src/quant_platform_kit/strategy_lifecycle/param_optimizer.py

@@ -234,10 +234,13 @@ def _attach_walkforward(
         benchmark_cagr=best_result.benchmark_cagr,
         benchmark_max_drawdown=best_result.benchmark_max_drawdown,
         excess_cagr=best_result.excess_cagr,
-        oos_sharpe=oos_sharpe, oos_calmar=oos_calmar, oos_max_dd=oos_max_dd,
+        oos_sharpe=oos_sharpe, oos_calmar=oos_calmar, oos_max_drawdown=oos_max_dd,
         walk_forward_stability=stability,
         run_id=best_result.run_id, run_duration_seconds=best_result.run_duration_seconds,
         source_script=best_result.source_script, computed_at=best_result.computed_at,
+        execution_timing=best_result.execution_timing,
+        result_identity_version=best_result.result_identity_version,
+        persist_mode=best_result.persist_mode,
     )
 
 

src/quant_platform_kit/strategy_lifecycle/performance_store.py

@@ -15,6 +15,7 @@
 
 import json
 import tempfile
+import time
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import date, datetime, timezone
@@ -34,12 +35,22 @@
 SCHEMA_VERSION = "strategy_lifecycle.v1"
 DEFAULT_BUCKET_ENV = "LIFECYCLE_PERFORMANCE_BUCKET"
 DEFAULT_LOCAL_ROOT = Path(tempfile.gettempdir()) / "quant_platform_lifecycle"
+LATEST_EXECUTION_TIMING = object()
+LEGACY_EXECUTION_TIMING = object()
 
 
 def _now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def _validate_execution_timing(value: Any) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str) or not value:
+        raise ValueError("execution_timing must be None or a non-empty string")
+    return value
+
+
 def _clean_key(value: str) -> str:
     parts = []
     for raw_part in str(value or "").replace("\\", "/").split("/"):
@@ -254,29 +265,52 @@ def load_latest_drift(self, domain: str, strategy_profile: str) -> DriftResult |
 
     def _backtest_key(self, result: BacktestResult) -> str:
         stamp = _clean_key(result.computed_at or result.run_id or result.param_set_id or _now_iso()).replace("/", "_")
+        timing = "t0_missing" if result.execution_timing is None else f"t1_{str(result.execution_timing).encode().hex()}"
         return (
             f"backtest/{_clean_key(result.domain)}/{_clean_key(result.strategy_profile)}/"
-            f"backtest_v{result.param_version}_{stamp}.json"
+            f"backtest_v{result.param_version}_{timing}_id{result.result_identity_version}_{stamp}.json"
         )
 
     def save_backtest_result(self, result: BacktestResult) -> None:
+        if result.persist_mode != "durable":
+            raise ValueError("PerformanceStore only supports durable backtest persistence")
+        _validate_execution_timing(result.execution_timing)
+        write_revision = time.time_ns()
         self._write(
             self._backtest_key(result),
-            {**result.to_dict(), "schema_version": SCHEMA_VERSION},
+            {
+                **result.to_dict(),
+                "schema_version": SCHEMA_VERSION,
+                "store_write_timestamp": _now_iso(),
+                "store_write_revision": str(write_revision),
+            },
         )
 
-    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 | object = LATEST_EXECUTION_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:
             return None
-        candidates: list[tuple[tuple[str, int, str], BacktestResult]] = []
+        candidates: list[tuple[tuple[str, int, str, str], BacktestResult]] = []
         for key in keys:
             data = self._read(key)
             result = _backtest_from_dict(data) if data else None
             if result is None:
                 continue
-            candidates.append((_backtest_sort_key(result, key), result))
+            write_revision = str(data.get("store_write_revision", "")) if data else ""
+            candidates.append((_backtest_sort_key(result, key, write_revision), result))
+        if not candidates:
+            return None
+        if execution_timing is LEGACY_EXECUTION_TIMING:
+            candidates = [item for item in candidates if item[1].execution_timing is None]
+        elif execution_timing is not LATEST_EXECUTION_TIMING and execution_timing is not None:
+            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])]
@@ -563,6 +597,13 @@ 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=(
+                _validate_execution_timing(data["execution_timing"])
+                if "execution_timing" in data
+                else None
+            ),
+            result_identity_version=int(data.get("result_identity_version", 1)),
+            persist_mode=str(data.get("persist_mode", "durable")),
             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,
@@ -597,9 +638,9 @@ def _backtest_from_dict(data: Mapping[str, Any]) -> BacktestResult | None:
         return None
 
 
-def _backtest_sort_key(result: BacktestResult, key: str) -> tuple[str, int, str]:
+def _backtest_sort_key(result: BacktestResult, key: str, write_revision: str = "") -> tuple[str, int, str, str]:
     computed_at = str(result.computed_at or "")
-    return (computed_at, int(result.param_version or 0), str(key))
+    return (computed_at, int(result.param_version or 0), str(write_revision), str(key))
 
 
 def _is_baseline_backtest(result: BacktestResult) -> bool:

tests/test_backtest_result_store_migration.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from datetime import date
+from pathlib import Path
+
+from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
+from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
+from quant_platform_kit.strategy_lifecycle.performance_store import (
+    LEGACY_EXECUTION_TIMING,
+    PerformanceStore,
+)
+
+
+def _result(*, timing: str | None, computed_at: str, profile: str = "SOXL") -> BacktestResult:
+    return BacktestResult(
+        strategy_profile=profile,
+        domain="us_equity",
+        param_set_id="baseline",
+        params={"lookback": 20},
+        execution_timing=timing,
+        result_identity_version=2,
+        persist_mode="durable",
+        computed_at=computed_at,
+        start_date=date(2020, 1, 1),
+        end_date=date(2024, 1, 1),
+    )
+
+
+class BacktestResultStoreMigrationTests(unittest.TestCase):
+    def test_old_positional_backtest_result_order_is_unchanged(self) -> None:
+        result = BacktestResult("s", "d", "p", {}, 3, 1.1, 0.9, 0.8, -0.2, 0.1, 0.3, 0.5, 0.4)
+        self.assertEqual(result.param_version, 3)
+        self.assertEqual(result.sharpe_ratio, 1.1)
+        self.assertEqual(result.total_return, 0.4)
+        self.assertIsNone(result.execution_timing)
+
+    def test_v1_record_without_metadata_remains_readable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            key = "backtest/us_equity/SOXL/backtest_v1_legacy.json"
+            store._write(key, {"strategy_profile": "SOXL", "domain": "us_equity", "param_set_id": "legacy", "params": {}})
+            loaded = store.load_latest_backtest("us_equity", "SOXL", execution_timing=LEGACY_EXECUTION_TIMING)
+        self.assertIsNotNone(loaded)
+        self.assertIsNone(loaded.execution_timing)
+        self.assertEqual(loaded.result_identity_version, 1)
+        self.assertEqual(loaded.persist_mode, "durable")
+
+    def test_mixed_timing_records_have_distinct_keys_and_filters(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store.save_backtest_result(_result(timing="next_open", computed_at="2026-01-01T00:00:00+00:00"))
+            store.save_backtest_result(_result(timing="next_close", computed_at="2026-01-01T00:00:00+00:00"))
+            keys = store._list_local_json_keys("backtest/us_equity/SOXL/")
+            latest = store.load_latest_backtest("us_equity", "SOXL")
+            opened = store.load_latest_backtest("us_equity", "SOXL", execution_timing="next_open")
+            closed = store.load_latest_backtest("us_equity", "SOXL", execution_timing="next_close")
+        self.assertEqual(len(keys), 2)
+        self.assertEqual(latest.execution_timing, "next_close")
+        self.assertEqual(opened.execution_timing, "next_open")
+        self.assertEqual(closed.execution_timing, "next_close")
+
+    def test_reserved_looking_timing_does_not_collide_with_missing_timing(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store.save_backtest_result(_result(timing=None, computed_at="2026-01-01T00:00:00+00:00"))
+            store.save_backtest_result(
+                _result(timing="legacy_unknown", computed_at="2026-01-01T00:00:00+00:00")
+            )
+            keys = store._list_local_json_keys("backtest/us_equity/SOXL/")
+        self.assertEqual(len(keys), 2)
+
+    def test_save_records_write_revision_for_equal_logical_slots(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store.save_backtest_result(_result(timing="next_open", computed_at="2026-01-01T00:00:00+00:00"))
+            store.save_backtest_result(_result(timing="next_close", computed_at="2026-01-01T00:00:00+00:00"))
+            payloads = [store._read(key) for key in store._list_local_json_keys("backtest/us_equity/SOXL/")]
+        revisions = [payload["store_write_revision"] for payload in payloads]
+        self.assertEqual(len(set(revisions)), 2)
+        self.assertTrue(all(payload.get("store_write_timestamp") for payload in payloads))
+
+    def test_explicit_legacy_sentinel_is_distinct_from_omitted_filter(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store.save_backtest_result(_result(timing=None, computed_at="2026-01-01T00:00:00+00:00"))
+            store.save_backtest_result(_result(timing="next_open", computed_at="2026-01-02T00:00:00+00:00"))
+            omitted = store.load_latest_backtest("us_equity", "SOXL")
+            explicit_none = store.load_latest_backtest("us_equity", "SOXL", execution_timing=None)
+            legacy = store.load_latest_backtest("us_equity", "SOXL", execution_timing=LEGACY_EXECUTION_TIMING)
+        self.assertEqual(omitted.execution_timing, "next_open")
+        self.assertEqual(explicit_none.execution_timing, "next_open")
+        self.assertIsNone(legacy.execution_timing)
+
+    def test_empty_timing_is_rejected_on_write_and_codec(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            with self.assertRaises(ValueError):
+                store.save_backtest_result(_result(timing="", computed_at="2026-01-01T00:00:00+00:00"))
+            store._write(
+                "backtest/us_equity/SOXL/invalid.json",
+                {"strategy_profile": "SOXL", "domain": "us_equity", "execution_timing": "", "params": {}},
+            )
+            self.assertIsNone(store.load_latest_backtest("us_equity", "SOXL", execution_timing=""))
+
+    def test_non_durable_persist_mode_is_rejected_before_write(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            result = _result(timing="next_open", computed_at="2026-01-01T00:00:00+00:00")
+            result = BacktestResult(**{**result.__dict__, "persist_mode": "ephemeral"})
+            with self.assertRaises(ValueError):
+                store.save_backtest_result(result)
+            self.assertEqual(list(Path(tmp).rglob("*.json")), [])
+
+    def test_orchestrator_persistence_preserves_result_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            orchestrator = BacktestOrchestrator(store=store)
+            source = _result(timing="next_close", computed_at="2026-01-01T00:00:00+00:00")
+            persisted = orchestrator.persist_result(
+                source,
+                strategy_profile=source.strategy_profile,
+                domain=source.domain,
+                params=source.params,
+                param_set_id=source.param_set_id,
+                param_version=source.param_version,
+            )
+            loaded = store.load_latest_backtest("us_equity", "SOXL", execution_timing="next_close")
+        self.assertEqual(persisted.execution_timing, "next_close")
+        self.assertEqual(persisted.result_identity_version, 2)
+        self.assertEqual(persisted.persist_mode, "durable")
+        self.assertEqual(loaded.execution_timing, "next_close")
+        self.assertEqual(loaded.result_identity_version, 2)
+
+    def test_result_json_round_trip_preserves_dates_and_nan(self) -> None:
+        result = _result(timing="next_open", computed_at="2026-01-01T00:00:00+00:00")
+        payload = result.to_dict()
+        self.assertEqual(payload["start_date"], "2020-01-01")
+        self.assertEqual(payload["end_date"], "2024-01-01")
+        self.assertEqual(json.loads(json.dumps(payload))["execution_timing"], "next_open")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_param_optimizer.py

@@ -1,14 +1,29 @@
 from __future__ import annotations
 
 import unittest
+from datetime import date
 from types import SimpleNamespace
 from unittest.mock import patch
 
 from quant_platform_kit.strategy_lifecycle.backtest_orchestrator import BacktestOrchestrator
-from quant_platform_kit.strategy_lifecycle.param_optimizer import _auto_register_runner
+from quant_platform_kit.strategy_lifecycle.contracts import BacktestResult
+from quant_platform_kit.strategy_lifecycle.param_optimizer import _attach_walkforward, _auto_register_runner
 
 
 class ParamOptimizerRunnerRegistrationTests(unittest.TestCase):
+    def test_attach_walkforward_preserves_result_identity_metadata(self) -> None:
+        result = BacktestResult(
+            strategy_profile="SOXL", domain="us_equity", param_set_id="grid",
+            params={"lookback": 20}, execution_timing="next_close",
+            result_identity_version=2, persist_mode="durable",
+            start_date=date(2020, 1, 1), end_date=date(2024, 1, 1), computed_at="2026-01-01",
+        )
+        with patch("quant_platform_kit.strategy_lifecycle.param_optimizer._run_walkforward_validation", return_value=(0.8, 1.0, 0.5, -0.2)):
+            copied = _attach_walkforward("SOXL", "us_equity", result, result.params, BacktestOrchestrator(), None, None)
+        self.assertEqual(copied.execution_timing, "next_close")
+        self.assertEqual(copied.result_identity_version, 2)
+        self.assertEqual(copied.persist_mode, "durable")
+
     def test_auto_register_runner_rejects_placeholder_runner(self) -> None:
         orchestrator = BacktestOrchestrator()