From db366f157c1c59a55cbc739d87cc23c755152b72 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:37:06 +0800 Subject: [PATCH 1/4] feat: add legacy-safe timing store codec Co-Authored-By: Codex --- .../strategy_lifecycle/contracts.py | 5 + .../strategy_lifecycle/performance_store.py | 60 ++++++++++-- tests/test_result_store_redesign_a.py | 97 +++++++++++++++++++ 3 files changed, 153 insertions(+), 9 deletions(-) create mode 100644 tests/test_result_store_redesign_a.py diff --git a/src/quant_platform_kit/strategy_lifecycle/contracts.py b/src/quant_platform_kit/strategy_lifecycle/contracts.py index dfb7921..6a84809 100644 --- a/src/quant_platform_kit/strategy_lifecycle/contracts.py +++ b/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, } diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 4fbe735..4fae1f4 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/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 @@ -30,10 +31,13 @@ StrategyPerformanceSnapshot, UpdateLogEntry, ) +from quant_platform_kit.strategy_lifecycle.capabilities import canonical_profile_id 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 +56,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 not isinstance(value, str) or not value: + raise ValueError("execution_timing must be None or a non-empty string") + return value + + @dataclass(frozen=True) class PerformanceStore: """Read/write strategy lifecycle data to local filesystem and/or cloud storage.""" @@ -254,29 +273,47 @@ 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, + "store_write_timestamp": _now_iso(), + "store_write_revision": str(time.time_ns()), + }, ) - def load_latest_backtest(self, domain: str, strategy_profile: str) -> BacktestResult | None: - prefix = f"backtest/{_clean_key(domain)}/{_clean_key(strategy_profile)}/" + def load_latest_backtest( + self, domain: str, strategy_profile: str, *, + execution_timing: str | object | None = LEGACY_EXECUTION_TIMING, + ) -> BacktestResult | None: + prefix = f"backtest/{_clean_key(domain)}/{_clean_key(_canonical_profile(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)) + revision = str(data.get("store_write_revision", "")) if data else "" + candidates.append((_backtest_sort_key(result, key, revision), 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])] @@ -563,6 +600,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 +639,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, 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(revision), str(key)) def _is_baseline_backtest(result: BacktestResult) -> bool: diff --git a/tests/test_result_store_redesign_a.py b/tests/test_result_store_redesign_a.py new file mode 100644 index 0000000..fa308ee --- /dev/null +++ b/tests/test_result_store_redesign_a.py @@ -0,0 +1,97 @@ +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.assertEqual(latest.execution_timing, "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)) + + +if __name__ == "__main__": + unittest.main() From 779be7ce5d22f80426637cdc183963908d1ffc4f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:44:47 +0800 Subject: [PATCH 2/4] fix: read legacy raw timing prefixes Co-Authored-By: Codex --- .../strategy_lifecycle/performance_store.py | 13 +++++++++-- tests/test_result_store_redesign_a.py | 22 +++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 4fae1f4..e85fff1 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -296,8 +296,17 @@ def load_latest_backtest( self, domain: str, strategy_profile: str, *, execution_timing: str | object | None = LEGACY_EXECUTION_TIMING, ) -> BacktestResult | None: - prefix = f"backtest/{_clean_key(domain)}/{_clean_key(_canonical_profile(strategy_profile))}/" - keys = list(dict.fromkeys([*self._list_cloud_keys(prefix), *self._list_local_json_keys(prefix)])) + canonical_profile = _clean_key(_canonical_profile(strategy_profile)) + raw_profile = _clean_key(strategy_profile) + prefixes = list(dict.fromkeys( + f"backtest/{_clean_key(domain)}/{profile}/" + for profile in (canonical_profile, raw_profile) + )) + 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, str], BacktestResult]] = [] diff --git a/tests/test_result_store_redesign_a.py b/tests/test_result_store_redesign_a.py index fa308ee..1d7a78c 100644 --- a/tests/test_result_store_redesign_a.py +++ b/tests/test_result_store_redesign_a.py @@ -92,6 +92,28 @@ def test_literal_empty_and_invalid_timing_fail_closed(self) -> None: ) 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_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() From a8738cc9ded00a29cf7ff9fe9d0466f01a5364a7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:54:12 +0800 Subject: [PATCH 3/4] fix: finalize legacy-safe timing selection Co-Authored-By: Codex --- .../strategy_lifecycle/capabilities.py | 20 +++++++---- .../strategy_lifecycle/performance_store.py | 28 +++++++-------- tests/test_result_store_redesign_a.py | 34 ++++++++++++++++++- 3 files changed, 60 insertions(+), 22 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/capabilities.py b/src/quant_platform_kit/strategy_lifecycle/capabilities.py index ddf0ca3..a865445 100644 --- a/src/quant_platform_kit/strategy_lifecycle/capabilities.py +++ b/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 { diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index e85fff1..36578f8 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -15,7 +15,6 @@ import json import tempfile -import time from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import date, datetime, timezone @@ -31,7 +30,7 @@ StrategyPerformanceSnapshot, UpdateLogEntry, ) -from quant_platform_kit.strategy_lifecycle.capabilities import canonical_profile_id +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" @@ -66,8 +65,8 @@ def _canonical_profile(value: str) -> str: 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") + if value not in {"next_open", "next_close"}: + raise ValueError("execution_timing must be None, next_open, or next_close") return value @@ -287,8 +286,6 @@ def save_backtest_result(self, result: BacktestResult) -> None: **result.to_dict(), "canonical_profile_id": _canonical_profile(result.strategy_profile), "schema_version": SCHEMA_VERSION, - "store_write_timestamp": _now_iso(), - "store_write_revision": str(time.time_ns()), }, ) @@ -296,11 +293,13 @@ def load_latest_backtest( self, domain: str, strategy_profile: str, *, execution_timing: str | object | None = LEGACY_EXECUTION_TIMING, ) -> BacktestResult | None: - canonical_profile = _clean_key(_canonical_profile(strategy_profile)) - raw_profile = _clean_key(strategy_profile) + try: + profiles = known_profile_aliases(strategy_profile) + except Exception: + profiles = (strategy_profile,) prefixes = list(dict.fromkeys( - f"backtest/{_clean_key(domain)}/{profile}/" - for profile in (canonical_profile, raw_profile) + f"backtest/{_clean_key(domain)}/{_clean_key(profile)}/" + for profile in profiles )) keys: list[str] = [] for prefix in prefixes: @@ -309,14 +308,13 @@ def load_latest_backtest( keys = list(dict.fromkeys(keys)) if not keys: return None - candidates: list[tuple[tuple[str, int, str, 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 if result is None: continue - revision = str(data.get("store_write_revision", "")) if data else "" - candidates.append((_backtest_sort_key(result, key, revision), result)) + 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: @@ -648,9 +646,9 @@ def _backtest_from_dict(data: Mapping[str, Any]) -> BacktestResult | None: return None -def _backtest_sort_key(result: BacktestResult, key: str, revision: str = "") -> tuple[str, int, str, 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(revision), 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: diff --git a/tests/test_result_store_redesign_a.py b/tests/test_result_store_redesign_a.py index 1d7a78c..daeb580 100644 --- a/tests/test_result_store_redesign_a.py +++ b/tests/test_result_store_redesign_a.py @@ -79,7 +79,7 @@ def test_mixed_timing_same_slot_uses_revision_only_for_any_timing(self) -> None: 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.assertEqual(latest.execution_timing, "next_close") + 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: @@ -104,6 +104,38 @@ def test_reads_legacy_raw_profile_prefixes_and_deduplicates_prefixes(self) -> No 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_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)) From 3d875bdf6e73779e534587e2d6c4448462229e06 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 10:13:35 +0800 Subject: [PATCH 4/4] fix: include exact legacy profile prefix Co-Authored-By: Codex --- .../strategy_lifecycle/performance_store.py | 1 + tests/test_result_store_redesign_a.py | 11 +++++++++++ 2 files changed, 12 insertions(+) diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_store.py b/src/quant_platform_kit/strategy_lifecycle/performance_store.py index 36578f8..d331461 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_store.py @@ -297,6 +297,7 @@ def load_latest_backtest( 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 diff --git a/tests/test_result_store_redesign_a.py b/tests/test_result_store_redesign_a.py index daeb580..efcbb08 100644 --- a/tests/test_result_store_redesign_a.py +++ b/tests/test_result_store_redesign_a.py @@ -115,6 +115,17 @@ def test_all_registered_alias_casings_are_discovered(self) -> None: 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))