feat: add explicit R1-S2 capability contracts

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,
+    }

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