diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py new file mode 100644 index 0000000..7b8242b --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py @@ -0,0 +1,226 @@ +"""Strict side-by-side vNext result/store contract foundation. + +Pure validation and deterministic wire/key generation only; this module performs +no store, cache, publish, runner, or filesystem I/O and never reads legacy data. + +Identity subset (and only this subset) feeds ``identity_digest`` and ``key``: +namespace, schema_version, domain, canonical_profile, execution_timing, +result_identity_version, strategy_id, run_id, param_set_id, param_version, +source_revision, and canonical params. ``persist_mode`` and ``computed_at`` are +wire metadata and are intentionally excluded from identity. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from datetime import datetime +from dataclasses import dataclass +from enum import Enum +from types import MappingProxyType +from typing import Any, Mapping + +from quant_platform_kit.strategy_lifecycle.capabilities import ExecutionTiming, PersistMode + +VNEXT_NAMESPACE = "qpk-vnext/result/v1" +_SCHEMA_VERSION = "qpk.vnext.result.v1" +_MAX_SEGMENT = 100 +MAX_SAFE_JSON_INTEGER = 2**53 - 1 +_RFC3339_UTC = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$") +_WIRE_FIELDS = frozenset({ + "namespace", "schema_version", "domain", "canonical_profile", "execution_timing", + "result_identity_version", "persist_mode", "strategy_id", "run_id", "param_set_id", + "param_version", "computed_at", "source_revision", "params", "identity_digest", +}) + + +class VNextContractError(ValueError): + """Sanitized validation error for all vNext contract failures.""" + + def __init__(self) -> None: + super().__init__("invalid vNext result/store contract") + + +def _segment(value: Any) -> str: + if not isinstance(value, str) or not value or len(value) > _MAX_SEGMENT: + raise VNextContractError() + if value in {".", ".."} or "/" in value or "\\" in value or value.startswith("/"): + raise VNextContractError() + if value != value.strip() or value != value.strip("-._"): + raise VNextContractError() + if any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for char in value): + raise VNextContractError() + return value + + +def _param_value(value: Any) -> Any: + if value is None: + raise VNextContractError() + if isinstance(value, (str, bool)): + return value + if isinstance(value, int): + if abs(value) > MAX_SAFE_JSON_INTEGER: + raise VNextContractError() + return value + if isinstance(value, float) and math.isfinite(value): + return value + if isinstance(value, tuple): + return tuple(_param_value(item) for item in value) + raise VNextContractError() + + +def _params(value: Any) -> dict[str, Any]: + if not isinstance(value, Mapping) or any(not isinstance(key, str) or not key for key in value): + raise VNextContractError() + return {key: _param_value(value[key]) for key in sorted(value)} + + +def _wire_params(value: Any) -> Mapping[str, Any]: + if not isinstance(value, Mapping) or any(not isinstance(key, str) or not key for key in value): + raise VNextContractError() + def thaw(item: Any) -> Any: + if isinstance(item, list): + return tuple(thaw(child) for child in item) + if isinstance(item, tuple): + return tuple(thaw(child) for child in item) + if isinstance(item, (str, bool)) or (isinstance(item, float) and math.isfinite(item)): + return item + if isinstance(item, int) and abs(item) <= MAX_SAFE_JSON_INTEGER: + return item + raise VNextContractError() + return {key: thaw(value[key]) for key in sorted(value)} + + +def _text(value: Any) -> str: + if not isinstance(value, str) or not value or len(value) > _MAX_SEGMENT: + raise VNextContractError() + if any(ord(char) < 32 or char in "/\\" for char in value): + raise VNextContractError() + return value + + +def _canonical_profile(value: Any) -> str: + """Validate an uppercase canonical profile; policy allowlists belong upstream.""" + value = _segment(value) + if value != value.upper(): + raise VNextContractError() + return value + + +def _computed_at(value: Any) -> str: + """Validate canonical UTC RFC3339 audit time without timezone conversion.""" + value = _text(value) + if not _RFC3339_UTC.fullmatch(value): + raise VNextContractError() + try: + datetime.fromisoformat(value.removesuffix("Z")) + except ValueError as exc: + raise VNextContractError() from exc + return value + + +def _canonical_json(value: Mapping[str, Any]) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + + +def _digest(value: Mapping[str, Any]) -> str: + return hashlib.sha256(_canonical_json(value)).hexdigest() + + +def _thaw_params(value: Mapping[str, Any]) -> dict[str, Any]: + def thaw(item: Any) -> Any: + if isinstance(item, tuple): + return [thaw(child) for child in item] + return item + return {key: thaw(value[key]) for key in sorted(value)} + + +@dataclass(frozen=True) +class VNextResultContract: + domain: str + canonical_profile: str + execution_timing: ExecutionTiming + result_identity_version: int + persist_mode: PersistMode + strategy_id: str + run_id: str + param_set_id: str + param_version: int + computed_at: str + source_revision: str + params: Mapping[str, Any] + + def __post_init__(self) -> None: + _segment(self.domain) + _canonical_profile(self.canonical_profile) + try: + object.__setattr__(self, "execution_timing", ExecutionTiming(self.execution_timing)) + object.__setattr__(self, "persist_mode", PersistMode(self.persist_mode)) + except Exception as exc: + raise VNextContractError() from exc + for value in (self.strategy_id, self.run_id, self.param_set_id, self.source_revision): + _segment(value) + if not isinstance(self.result_identity_version, int) or isinstance(self.result_identity_version, bool) or self.result_identity_version < 1: + raise VNextContractError() + if not isinstance(self.param_version, int) or isinstance(self.param_version, bool) or self.param_version < 1: + raise VNextContractError() + _computed_at(self.computed_at) + object.__setattr__(self, "params", MappingProxyType(_params(self.params))) + + def _identity_payload(self) -> dict[str, Any]: + return { + "namespace": VNEXT_NAMESPACE, + "schema_version": _SCHEMA_VERSION, + "domain": self.domain, + "canonical_profile": self.canonical_profile, + "execution_timing": self.execution_timing.value, + "result_identity_version": self.result_identity_version, + "strategy_id": self.strategy_id, + "run_id": self.run_id, + "param_set_id": self.param_set_id, + "param_version": self.param_version, + "source_revision": self.source_revision, + "params": _thaw_params(self.params), + } + + def _payload(self) -> dict[str, Any]: + return { + **self._identity_payload(), + "persist_mode": self.persist_mode.value, + "computed_at": self.computed_at, + } + + def to_wire(self) -> dict[str, Any]: + payload = self._payload() + return {**payload, "identity_digest": _digest(self._identity_payload())} + + @property + def key(self) -> str: + return f"{VNEXT_NAMESPACE}/{self.domain}/{self.canonical_profile}/{self.strategy_id}/{self.run_id}/{self.execution_timing.value}/i{self.result_identity_version}/p{self.param_version}/{_digest(self._identity_payload())}.json" + + +def decode_vnext_wire(data: Mapping[str, Any]) -> VNextResultContract: + try: + if not isinstance(data, Mapping) or set(data) != _WIRE_FIELDS: + raise VNextContractError() + if data["namespace"] != VNEXT_NAMESPACE or data["schema_version"] != _SCHEMA_VERSION: + raise VNextContractError() + wire_params = _wire_params(data["params"]) + contract = VNextResultContract( + domain=data["domain"], canonical_profile=data["canonical_profile"], + execution_timing=data["execution_timing"], result_identity_version=data["result_identity_version"], + persist_mode=data["persist_mode"], strategy_id=data["strategy_id"], run_id=data["run_id"], + param_set_id=data["param_set_id"], param_version=data["param_version"], + computed_at=data["computed_at"], source_revision=data["source_revision"], params=wire_params, + ) + if not isinstance(data["identity_digest"], str) or _digest(contract._identity_payload()) != data["identity_digest"]: + raise VNextContractError() + if contract.to_wire() != dict(data): + raise VNextContractError() + return contract + except VNextContractError: + raise + except Exception as exc: + raise VNextContractError() from exc diff --git a/tests/test_vnext_result_store_contract.py b/tests/test_vnext_result_store_contract.py new file mode 100644 index 0000000..41e3ad2 --- /dev/null +++ b/tests/test_vnext_result_store_contract.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import copy +import unittest +from dataclasses import replace + +from quant_platform_kit.strategy_lifecycle.capabilities import ExecutionTiming, PersistMode +from quant_platform_kit.strategy_lifecycle.vnext_result_store_contract import ( + MAX_SAFE_JSON_INTEGER, + VNEXT_NAMESPACE, + VNextContractError, + VNextResultContract, + decode_vnext_wire, +) + + +def _contract(**overrides): + values = { + "domain": "us_equity", + "canonical_profile": "SOXL", + "execution_timing": ExecutionTiming.NEXT_OPEN, + "result_identity_version": 1, + "persist_mode": PersistMode.DURABLE, + "strategy_id": "soxl_trend", + "run_id": "run-001", + "param_set_id": "baseline", + "param_version": 1, + "computed_at": "2026-07-15T00:00:00Z", + "source_revision": "abc123", + "params": {"lookback": 20, "threshold": 0.5}, + } + values.update(overrides) + return VNextResultContract(**values) + + +class VNextResultStoreContractTests(unittest.TestCase): + def test_round_trip_and_namespace_isolation(self) -> None: + contract = _contract() + wire = contract.to_wire() + self.assertEqual(wire["namespace"], VNEXT_NAMESPACE) + self.assertEqual(decode_vnext_wire(wire), contract) + self.assertTrue(contract.key.startswith("qpk-vnext/result/v1/")) + upro = _contract(canonical_profile="UPRO") + self.assertEqual(decode_vnext_wire(upro.to_wire()), upro) + + def test_order_independence_and_identity_changes(self) -> None: + first = _contract(params={"b": 2, "a": 1}) + second = _contract(params={"a": 1, "b": 2}) + self.assertEqual(first.to_wire(), second.to_wire()) + self.assertEqual(first.key, second.key) + self.assertNotEqual(first.key, replace(first, run_id="run-002").key) + + def test_wire_metadata_does_not_change_identity(self) -> None: + first = _contract(params={"nested": (1, (2, 3))}) + second = replace(first, persist_mode=PersistMode.EPHEMERAL, computed_at="2026-07-16T00:00:00Z") + self.assertEqual(first.key, second.key) + self.assertEqual(first.to_wire()["identity_digest"], second.to_wire()["identity_digest"]) + self.assertNotEqual(first.to_wire()["persist_mode"], second.to_wire()["persist_mode"]) + self.assertNotEqual(first.to_wire()["computed_at"], second.to_wire()["computed_at"]) + + def test_identity_fields_change_digest_and_key(self) -> None: + base = _contract() + variants = ( + replace(base, execution_timing=ExecutionTiming.NEXT_CLOSE), + replace(base, canonical_profile="TQQQ"), + replace(base, source_revision="def456"), + replace(base, param_set_id="candidate"), + replace(base, result_identity_version=2), + ) + for variant in variants: + self.assertNotEqual(base.key, variant.key) + self.assertNotEqual(base.to_wire()["identity_digest"], variant.to_wire()["identity_digest"]) + + def test_required_none_legacy_any_alias_and_unknown_values_reject(self) -> None: + for field in ("execution_timing", "canonical_profile", "strategy_id", "run_id", "source_revision"): + with self.assertRaises(VNextContractError): + _contract(**{field: None}) + with self.assertRaises(VNextContractError): + _contract(execution_timing=None) + with self.assertRaises(VNextContractError): + _contract(canonical_profile="soxl_soxx_trend_income") + for profile in ("", "soxl", "A/B", "..", "X" * 101): + with self.assertRaises(VNextContractError): + _contract(canonical_profile=profile) + with self.assertRaises(VNextContractError): + _contract(execution_timing="ANY") + + def test_persist_modes_are_explicit_without_io(self) -> None: + self.assertEqual(_contract(persist_mode=PersistMode.EPHEMERAL).to_wire()["persist_mode"], "ephemeral") + self.assertEqual(_contract(persist_mode=PersistMode.DURABLE).to_wire()["persist_mode"], "durable") + + def test_tuple_and_nested_tuple_round_trip_freezes_params(self) -> None: + contract = _contract(params={"levels": (1, ("a", True))}) + self.assertEqual(decode_vnext_wire(contract.to_wire()), contract) + self.assertIsInstance(contract.params["levels"], tuple) + with self.assertRaises(AttributeError): + contract.params["levels"].append(2) + + def test_direct_mutable_values_reject_but_wire_lists_are_frozen(self) -> None: + with self.assertRaises(VNextContractError): + _contract(params={"values": [1, 2]}) + decoded = decode_vnext_wire(_contract(params={"values": (1, 2)}).to_wire()) + self.assertEqual(decoded.params["values"], (1, 2)) + + def test_json_safe_integer_boundary_and_nested_rejection(self) -> None: + boundary = _contract(params={"low": -MAX_SAFE_JSON_INTEGER, "high": MAX_SAFE_JSON_INTEGER}) + self.assertEqual(decode_vnext_wire(boundary.to_wire()), boundary) + self.assertEqual(_contract(params={"flag": True}).params["flag"], True) + for value in (MAX_SAFE_JSON_INTEGER + 1, -(MAX_SAFE_JSON_INTEGER + 1)): + with self.assertRaises(VNextContractError): + _contract(params={"value": value}) + wire = _contract(params={"value": (1,)}).to_wire() + wire["params"] = {"nested": [value]} + with self.assertRaises(VNextContractError): + decode_vnext_wire(wire) + + def test_computed_at_requires_canonical_utc_rfc3339(self) -> None: + for timestamp in ( + "", " 2026-07-15T00:00:00Z", "2026-07-15T00:00:00+00:00", + "2026-02-30T00:00:00Z", "2026-07-15", "not-a-time", + ): + with self.assertRaises(VNextContractError): + _contract(computed_at=timestamp) + for timestamp in ("2026-07-15T00:00:00Z", "2026-07-15T00:00:00.123456Z"): + contract = _contract(computed_at=timestamp) + self.assertEqual(decode_vnext_wire(contract.to_wire()), contract) + + def test_adversarial_wire_values_reject(self) -> None: + wire = _contract().to_wire() + for mutation in ( + {"namespace": "strategy_lifecycle"}, + {"execution_timing": None}, + {"canonical_profile": "SOXL_SOXX_TREND_INCOME"}, + {"result_identity_version": True}, + {"params": {"bad": []}}, + {"unknown": "field"}, + ): + candidate = copy.deepcopy(wire) + candidate.update(mutation) + with self.assertRaises(VNextContractError): + decode_vnext_wire(candidate) + missing = copy.deepcopy(wire) + del missing["run_id"] + with self.assertRaises(VNextContractError): + decode_vnext_wire(missing) + + def test_unsafe_and_overlong_identity_segments_reject(self) -> None: + for field in ("domain", "strategy_id", "run_id", "param_set_id", "source_revision"): + with self.assertRaises(VNextContractError): + _contract(**{field: "../unsafe"}) + with self.assertRaises(VNextContractError): + _contract(**{field: "x" * 101}) + with self.assertRaises(VNextContractError): + _contract(param_version=True) + + +if __name__ == "__main__": + unittest.main()