feat: legacy-safe result/store timing codec

src/quant_platform_kit/strategy_lifecycle/capabilities.py

@@ -103,20 +103,28 @@ def validate_capability(request: BacktestRequest, capabilities: BacktestCapabili
         raise CapabilityError(f"persist_mode={request.persist_mode.value} is not supported")
 
 
+PROFILE_ALIASES: dict[str, tuple[str, ...]] = {
+    "SOXL": ("SOXL", "SOXL_SOXX_TREND_INCOME"),
+    "TQQQ": ("TQQQ", "TQQQ_GROWTH_INCOME"),
+}
+
+
 def canonical_profile_id(profile: str) -> str:
     """Pure fixture normalization for the two R1 profiles; no runtime routing."""
-    aliases = {
-        "SOXL": "SOXL",
-        "SOXL_SOXX_TREND_INCOME": "SOXL",
-        "TQQQ": "TQQQ",
-        "TQQQ_GROWTH_INCOME": "TQQQ",
-    }
+    aliases = {alias: canonical for canonical, values in PROFILE_ALIASES.items() for alias in values}
     try:
         return aliases[str(profile).strip().upper()]
     except KeyError as exc:
         raise CapabilityError(f"unsupported canonical profile: {profile}") from exc
 
 
+def known_profile_aliases(profile: str) -> tuple[str, ...]:
+    """Return canonical and registered legacy aliases with common casings."""
+    canonical = canonical_profile_id(profile)
+    values = PROFILE_ALIASES[canonical]
+    return tuple(dict.fromkeys(alias_case for alias in values for alias_case in (alias, alias.lower(), alias.upper())))
+
+
 def serialize_request(request: BacktestRequest) -> dict[str, Any]:
     """Return a deterministic, side-effect-free contract fixture."""
     return {

src/quant_platform_kit/strategy_lifecycle/contracts.py

@@ -263,6 +263,9 @@ class BacktestResult:
     source_revision: str = ""
     cost_model: str = ""
 
+    execution_timing: str | None = field(default=None, kw_only=True)
+    result_identity_version: int = field(default=2, kw_only=True)
+
     def to_dict(self) -> dict[str, object]:
         return {
             "strategy_profile": self.strategy_profile,
@@ -295,6 +298,8 @@ 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,
         }
 
 

src/quant_platform_kit/strategy_lifecycle/performance_store.py

@@ -30,10 +30,13 @@
     StrategyPerformanceSnapshot,
     UpdateLogEntry,
 )
+from quant_platform_kit.strategy_lifecycle.capabilities import canonical_profile_id, known_profile_aliases
 
 SCHEMA_VERSION = "strategy_lifecycle.v1"
 DEFAULT_BUCKET_ENV = "LIFECYCLE_PERFORMANCE_BUCKET"
 DEFAULT_LOCAL_ROOT = Path(tempfile.gettempdir()) / "quant_platform_lifecycle"
+ANY_EXECUTION_TIMING = object()
+LEGACY_EXECUTION_TIMING = object()
 
 
 def _now_iso() -> str:
@@ -52,6 +55,21 @@ def _clean_key(value: str) -> str:
     return "/".join(parts) or "unknown"
 
 
+def _canonical_profile(value: str) -> str:
+    try:
+        return canonical_profile_id(value)
+    except Exception:
+        return str(value)
+
+
+def _validate_execution_timing(value: Any) -> str | None:
+    if value is None:
+        return None
+    if value not in {"next_open", "next_close"}:
+        raise ValueError("execution_timing must be None, next_open, or next_close")
+    return value
+
+
 @dataclass(frozen=True)
 class PerformanceStore:
     """Read/write strategy lifecycle data to local filesystem and/or cloud storage."""
@@ -254,23 +272,44 @@ 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_{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/{_clean_key(result.domain)}/{_clean_key(_canonical_profile(result.strategy_profile))}/"
+            f"backtest_v{result.param_version}_{timing}_id{result.result_identity_version}_{stamp}.json"
         )
 
     def save_backtest_result(self, result: BacktestResult) -> None:
+        _validate_execution_timing(result.execution_timing)
         self._write(
             self._backtest_key(result),
-            {**result.to_dict(), "schema_version": SCHEMA_VERSION},
+            {
+                **result.to_dict(),
+                "canonical_profile_id": _canonical_profile(result.strategy_profile),
+                "schema_version": SCHEMA_VERSION,
+            },
         )
 
-    def load_latest_backtest(self, domain: str, strategy_profile: str) -> 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)]))
+    def load_latest_backtest(
+        self, domain: str, strategy_profile: str, *,
+        execution_timing: str | object | None = LEGACY_EXECUTION_TIMING,
+    ) -> BacktestResult | None:
+        try:
+            profiles = known_profile_aliases(strategy_profile)
+        except Exception:
+            profiles = (strategy_profile,)
+        profiles = (*profiles, strategy_profile)
+        prefixes = list(dict.fromkeys(
+            f"backtest/{_clean_key(domain)}/{_clean_key(profile)}/"
+            for profile in profiles
+        ))
+        keys: list[str] = []
+        for prefix in prefixes:
+            keys.extend(self._list_cloud_keys(prefix))
+            keys.extend(self._list_local_json_keys(prefix))
+        keys = list(dict.fromkeys(keys))
         if not keys:
             return None
-        candidates: list[tuple[tuple[str, int, str], BacktestResult]] = []
+        candidates: list[tuple[tuple[str, int, int, str], BacktestResult]] = []
         for key in keys:
             data = self._read(key)
             result = _backtest_from_dict(data) if data else None
@@ -279,6 +318,12 @@ 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 LEGACY_EXECUTION_TIMING or execution_timing is None:
+            candidates = [item for item in candidates if item[1].execution_timing is None]
+        elif execution_timing is not ANY_EXECUTION_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 +608,11 @@ 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)),
             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 +647,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) -> tuple[str, int, int, 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), int(result.result_identity_version or 0), str(key))
 
 
 def _is_baseline_backtest(result: BacktestResult) -> bool:

tests/test_result_store_redesign_a.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+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.performance_store import (
+    ANY_EXECUTION_TIMING,
+    LEGACY_EXECUTION_TIMING,
+    PerformanceStore,
+)
+
+
+def _result(*, timing: str | None, computed_at: str = "2026-01-01T00:00:00+00:00", profile: str = "SOXL"):
+    return BacktestResult(
+        strategy_profile=profile,
+        domain="us_equity",
+        param_set_id="baseline",
+        params={"lookback": 20},
+        execution_timing=timing,
+        computed_at=computed_at,
+        start_date=date(2020, 1, 1),
+        end_date=date(2024, 1, 1),
+    )
+
+
+class ResultStoreRedesignATests(unittest.TestCase):
+    def test_old_positional_order_and_v1_codec(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.total_return, 0.4)
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store._write(
+                "backtest/us_equity/SOXL/backtest_v1_legacy.json",
+                {"strategy_profile": "SOXL", "domain": "us_equity", "param_set_id": "legacy", "params": {}},
+            )
+            loaded = store.load_latest_backtest("us_equity", "SOXL")
+        self.assertIsNotNone(loaded)
+        self.assertIsNone(loaded.execution_timing)
+        self.assertEqual(loaded.result_identity_version, 1)
+
+    def test_timing_round_trip_and_aliases_have_same_key(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            opened = _result(timing="next_open", profile="SOXL")
+            alias = _result(timing="next_open", profile="soxl_soxx_trend_income")
+            self.assertEqual(store._backtest_key(opened), store._backtest_key(alias))
+            store.save_backtest_result(opened)
+            payload = store._read(store._list_local_json_keys("backtest/us_equity/SOXL/")[0])
+            loaded = store.load_latest_backtest("us_equity", "SOXL", execution_timing="next_open")
+        self.assertEqual(payload["canonical_profile_id"], "SOXL")
+        self.assertEqual(loaded.execution_timing, "next_open")
+        self.assertEqual(loaded.result_identity_version, 2)
+
+    def test_default_none_legacy_any_and_specific_filters(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"))
+            default = 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)
+            any_timing = store.load_latest_backtest("us_equity", "SOXL", execution_timing=ANY_EXECUTION_TIMING)
+            specific = store.load_latest_backtest("us_equity", "SOXL", execution_timing="next_open")
+        self.assertIsNone(default.execution_timing)
+        self.assertIsNone(explicit_none.execution_timing)
+        self.assertIsNone(legacy.execution_timing)
+        self.assertEqual(any_timing.execution_timing, "next_open")
+        self.assertEqual(specific.execution_timing, "next_open")
+
+    def test_mixed_timing_same_slot_uses_revision_only_for_any_timing(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store.save_backtest_result(_result(timing="next_open"))
+            store.save_backtest_result(_result(timing="next_close"))
+            latest = store.load_latest_backtest("us_equity", "SOXL", execution_timing=ANY_EXECUTION_TIMING)
+            legacy_key = "backtest/us_equity/SOXL/backtest_v1_legacy.json"
+            store._write(legacy_key, {"strategy_profile": "SOXL", "domain": "us_equity", "params": {}})
+            legacy = store.load_latest_backtest("us_equity", "SOXL", execution_timing=LEGACY_EXECUTION_TIMING)
+        self.assertIn(latest.execution_timing, {"next_open", "next_close"})
+        self.assertIsNone(legacy.execution_timing)
+
+    def test_literal_empty_and_invalid_timing_fail_closed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            for value in ("", 123):
+                store._write(
+                    f"backtest/us_equity/SOXL/invalid-{value}.json",
+                    {"strategy_profile": "SOXL", "domain": "us_equity", "execution_timing": value, "params": {}},
+                )
+            self.assertIsNone(store.load_latest_backtest("us_equity", "SOXL", execution_timing=ANY_EXECUTION_TIMING))
+
+    def test_reads_legacy_raw_profile_prefixes_and_deduplicates_prefixes(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            legacy = _result(timing=None, computed_at="2026-01-01T00:00:00+00:00", profile="soxl_soxx_trend_income")
+            raw_key = "backtest/us_equity/soxl_soxx_trend_income/backtest_v1_legacy.json"
+            store._write(raw_key, legacy.to_dict())
+            loaded_legacy = store.load_latest_backtest("us_equity", "soxl_soxx_trend_income")
+            store.save_backtest_result(_result(timing=None, computed_at="2026-01-02T00:00:00+00:00"))
+            loaded_mixed = store.load_latest_backtest("us_equity", "SOXL")
+        self.assertEqual(loaded_legacy.computed_at, "2026-01-01T00:00:00+00:00")
+        self.assertEqual(loaded_mixed.computed_at, "2026-01-02T00:00:00+00:00")
+
+    def test_all_registered_alias_casings_are_discovered(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            for index, alias in enumerate(("SOXL", "soxl", "SOXL_SOXX_TREND_INCOME", "soxl_soxx_trend_income")):
+                store._write(
+                    f"backtest/us_equity/{alias}/record-{index}.json",
+                    _result(timing=None, computed_at=f"2026-01-0{index + 1}T00:00:00+00:00").to_dict(),
+                )
+            loaded = store.load_latest_backtest("us_equity", "SOXL")
+        self.assertEqual(loaded.computed_at, "2026-01-04T00:00:00+00:00")
+
+    def test_exact_arbitrary_mixed_case_alias_prefix_is_discovered(self) -> None:
+        mixed_alias = "SoXl_SoXx_TrEnD_InCoMe"
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store._write(
+                f"backtest/us_equity/{mixed_alias}/record.json",
+                _result(timing=None, computed_at="2026-01-05T00:00:00+00:00", profile=mixed_alias).to_dict(),
+            )
+            loaded = store.load_latest_backtest("us_equity", mixed_alias)
+        self.assertEqual(loaded.computed_at, "2026-01-05T00:00:00+00:00")
+
+    def test_any_timing_is_stable_independent_of_save_order(self) -> None:
+        with tempfile.TemporaryDirectory() as first_tmp, tempfile.TemporaryDirectory() as second_tmp:
+            first = PerformanceStore(local_root=Path(first_tmp))
+            second = PerformanceStore(local_root=Path(second_tmp))
+            opened = _result(timing="next_open")
+            closed = _result(timing="next_close")
+            first.save_backtest_result(opened)
+            first.save_backtest_result(closed)
+            second.save_backtest_result(closed)
+            second.save_backtest_result(opened)
+            first_latest = first.load_latest_backtest("us_equity", "SOXL", execution_timing=ANY_EXECUTION_TIMING)
+            second_latest = second.load_latest_backtest("us_equity", "SOXL", execution_timing=ANY_EXECUTION_TIMING)
+        self.assertEqual(first_latest.execution_timing, second_latest.execution_timing)
+
+    def test_long_or_unknown_timing_is_rejected_on_write(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            for timing in ("next_open_" + "x" * 200, "legacy_unknown"):
+                with self.assertRaises(ValueError):
+                    store.save_backtest_result(_result(timing=timing))
+
+    def test_path_sanitization_applies_to_raw_prefix(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = PerformanceStore(local_root=Path(tmp))
+            store._write(
+                "backtest/us_equity/unsafe-profile/backtest_v1_legacy.json",
+                {"strategy_profile": "unsafe-profile", "domain": "us_equity", "params": {}},
+            )
+            loaded = store.load_latest_backtest("us_equity", "../unsafe-profile")
+        self.assertIsNotNone(loaded)
+
+
+if __name__ == "__main__":
+    unittest.main()