diff --git a/src/quant_platform_kit/strategy_lifecycle/capabilities.py b/src/quant_platform_kit/strategy_lifecycle/capabilities.py new file mode 100644 index 0000000..ddf0ca3 --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/capabilities.py @@ -0,0 +1,128 @@ +"""Pure R1-S2 capability/request contracts. + +This foundation is intentionally not wired into the runtime orchestrator. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +import math +from types import MappingProxyType +from typing import Any, Mapping + + +class CapabilityError(ValueError): + """Raised when a request is not explicitly supported by a capability contract.""" + + +class ExecutionTiming(str, Enum): + NEXT_OPEN = "next_open" + NEXT_CLOSE = "next_close" + + +class PersistMode(str, Enum): + DURABLE = "durable" + EPHEMERAL = "ephemeral" + + +def _freeze(value: Any) -> Any: + if isinstance(value, Mapping): + if any(not isinstance(key, str) for key in value): + raise CapabilityError("unsupported non-string parameter key") + return MappingProxyType({key: _freeze(item) for key, item in value.items()}) + if isinstance(value, list): + return tuple(_freeze(item) for item in value) + if isinstance(value, tuple): + return tuple(_freeze(item) for item in value) + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, float) and math.isfinite(value): + return value + raise CapabilityError(f"unsupported parameter value type: {type(value).__name__}") + + +def _thaw(value: Any) -> Any: + if isinstance(value, Mapping): + return {str(key): _thaw(item) for key, item in value.items()} + if isinstance(value, tuple): + return [_thaw(item) for item in value] + return value + + +@dataclass(frozen=True) +class BacktestCapabilities: + """Versioned, explicit capabilities declared by a runner adapter.""" + + contract_version: int = field(default=1, kw_only=True) + execution_timings: frozenset[ExecutionTiming] = field(default_factory=frozenset, kw_only=True) + ephemeral: bool = field(default=False, kw_only=True) + persist_modes: frozenset[PersistMode] = field( + default_factory=lambda: frozenset({PersistMode.DURABLE}), kw_only=True + ) + + +@dataclass(frozen=True) +class BacktestRequest: + """Immutable request used by a future capability-aware adapter boundary.""" + + profile: str = field(kw_only=True) + params: Mapping[str, Any] = field(kw_only=True) + execution_timing: ExecutionTiming | None = field(default=None, kw_only=True) + persist_mode: PersistMode = field(default=PersistMode.DURABLE, kw_only=True) + contract_version: int = field(default=1, kw_only=True) + + def __post_init__(self) -> None: + object.__setattr__(self, "profile", str(self.profile).strip()) + object.__setattr__(self, "params", _freeze(dict(self.params))) + object.__setattr__(self, "execution_timing", _coerce_timing(self.execution_timing)) + object.__setattr__(self, "persist_mode", PersistMode(self.persist_mode)) + + +class LegacyCapabilityAdapter: + """Pure declaration for legacy durable/no-timing runners.""" + + @staticmethod + def capabilities() -> BacktestCapabilities: + return BacktestCapabilities() + + +def _coerce_timing(value: ExecutionTiming | str | None) -> ExecutionTiming | None: + return None if value is None else ExecutionTiming(value) + + +def validate_capability(request: BacktestRequest, capabilities: BacktestCapabilities) -> None: + """Fail closed unless every requested semantic is explicitly declared.""" + if request.contract_version != capabilities.contract_version: + raise CapabilityError("contract_version is not supported") + if request.execution_timing is not None and request.execution_timing not in capabilities.execution_timings: + raise CapabilityError("execution_timing is not explicitly supported") + if request.persist_mode is PersistMode.EPHEMERAL and not capabilities.ephemeral: + raise CapabilityError("ephemeral execution is not explicitly supported") + if request.persist_mode is PersistMode.DURABLE and request.persist_mode not in capabilities.persist_modes: + raise CapabilityError(f"persist_mode={request.persist_mode.value} is not supported") + + +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", + } + try: + return aliases[str(profile).strip().upper()] + except KeyError as exc: + raise CapabilityError(f"unsupported canonical profile: {profile}") from exc + + +def serialize_request(request: BacktestRequest) -> dict[str, Any]: + """Return a deterministic, side-effect-free contract fixture.""" + return { + "contract_version": request.contract_version, + "profile": canonical_profile_id(request.profile), + "params": _thaw(request.params), + "execution_timing": request.execution_timing.value if request.execution_timing else None, + "persist_mode": request.persist_mode.value, + } diff --git a/tests/test_backtest_capabilities.py b/tests/test_backtest_capabilities.py new file mode 100644 index 0000000..5ec280e --- /dev/null +++ b/tests/test_backtest_capabilities.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from dataclasses import FrozenInstanceError + +import pytest + +from quant_platform_kit.strategy_lifecycle.capabilities import ( + BacktestCapabilities, + BacktestRequest, + CapabilityError, + ExecutionTiming, + LegacyCapabilityAdapter, + PersistMode, + canonical_profile_id, + serialize_request, + validate_capability, +) + + +def test_legacy_durable_without_timing_is_expressible() -> None: + request = BacktestRequest(profile="soxl_soxx_trend_income", params={"lookback": 20}) + capabilities = LegacyCapabilityAdapter.capabilities() + validate_capability(request, capabilities) + assert request.persist_mode is PersistMode.DURABLE + assert request.execution_timing is None + + +@pytest.mark.parametrize("timing", [ExecutionTiming.NEXT_OPEN, ExecutionTiming.NEXT_CLOSE]) +def test_explicit_timing_requires_declared_capability(timing: ExecutionTiming) -> None: + request = BacktestRequest(profile="SOXL", params={}, execution_timing=timing) + with pytest.raises(CapabilityError, match="execution_timing"): + validate_capability(request, LegacyCapabilityAdapter.capabilities()) + validate_capability(request, BacktestCapabilities(execution_timings=frozenset({timing}))) + + +def test_ephemeral_requires_explicit_capability() -> None: + request = BacktestRequest(profile="TQQQ", params={}, persist_mode=PersistMode.EPHEMERAL) + with pytest.raises(CapabilityError, match="ephemeral"): + validate_capability(request, LegacyCapabilityAdapter.capabilities()) + validate_capability(request, BacktestCapabilities(ephemeral=True)) + + +def test_version_mismatch_fails_closed_before_runner() -> None: + request = BacktestRequest(profile="TQQQ", params={}) + with pytest.raises(CapabilityError, match="contract_version"): + validate_capability(request, BacktestCapabilities(contract_version=2)) + + +def test_request_is_keyword_only_and_immutable() -> None: + request = BacktestRequest(profile="SOXL", params={"lookback": 20}) + with pytest.raises(TypeError): + BacktestRequest("SOXL", {}) + with pytest.raises(FrozenInstanceError): + request.profile = "TQQQ" + with pytest.raises(TypeError): + request.params["lookback"] = 30 + with pytest.raises(TypeError): + request.params["nested"] = {"x": 1} + + +def test_nested_mutable_values_are_frozen_and_deterministic() -> None: + left = BacktestRequest(profile="SOXL", params={"nested": {"items": [1, 2]}}) + right = BacktestRequest(profile="SOXL", params={"nested": {"items": [1, 2]}}) + assert serialize_request(left) == serialize_request(right) + with pytest.raises(TypeError): + left.params["nested"]["items"] += (3,) + + +@pytest.mark.parametrize("unsupported", [{"x"}, b"bytes", object()]) +def test_unsupported_parameter_values_are_rejected_at_construction(unsupported) -> None: + with pytest.raises(CapabilityError, match="unsupported"): + BacktestRequest(profile="SOXL", params={"value": unsupported}) + + +def test_supported_json_values_are_unchanged() -> None: + request = BacktestRequest(profile="SOXL", params={"none": None, "flag": True, "ratio": 0.5, "text": "ok"}) + assert serialize_request(request)["params"] == {"none": None, "flag": True, "ratio": 0.5, "text": "ok"} + + +def test_kwargs_and_signature_are_not_capability_evidence() -> None: + request = BacktestRequest(profile="SOXL", params={}, execution_timing=ExecutionTiming.NEXT_OPEN) + kwargs_like = BacktestCapabilities() + with pytest.raises(CapabilityError): + validate_capability(request, kwargs_like) + + +def test_canonical_ids_and_serialization_shape_are_pure() -> None: + request = BacktestRequest(profile="soxl_soxx_trend_income", params={"lookback": 20}) + assert canonical_profile_id(request.profile) == "SOXL" + expected = { + "contract_version": 1, + "profile": "SOXL", + "params": {"lookback": 20}, + "execution_timing": None, + "persist_mode": "durable", + } + assert serialize_request(request) == expected + assert serialize_request(BacktestRequest(profile="SOXL", params={"lookback": 20})) == expected