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/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..d331461 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_store.py +++ b/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: diff --git a/tests/test_result_store_redesign_a.py b/tests/test_result_store_redesign_a.py new file mode 100644 index 0000000..efcbb08 --- /dev/null +++ b/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()