feat: converged qpk-vnext N1 param grammar contract

docs/qpk_vnext_n1_param_grammar.md

@@ -0,0 +1,27 @@
+# QPK vNext N1 Params Grammar
+
+Approved clean-slate source of truth for `qpk-vnext/result/v2`:
+
+```text
+Params      := JSON object / safe Mapping<Name, Value> (top-level only)
+Name        := non-empty bounded Unicode-scalar string
+Value       := bool | safe integer | bounded Unicode-scalar string | Sequence
+Sequence    := tuple at Python construction; JSON array on wire; tuple after decode
+```
+
+`null`, float (including finite, `-0.0`, NaN, Inf), nested Mapping/object,
+mutable list/dict at Python construction, unknown types, unsafe/overlong strings,
+surrogates, and excessive recursion/size fail closed with sanitized contract errors.
+Decimal values must be canonical strings or scaled safe integers. Profile policy is
+owned by higher layers; this contract only requires an uppercase safe identifier.
+
+Every component emitted into the slash-delimited key is a single safe segment.
+This includes `strategy_id` and `run_id`: bounded `[A-Za-z0-9_.-]` text beginning
+with an alphanumeric character, with no slash, dot-segment, control character,
+or implicit encoding. `param_set_id` and `source_revision` remain identity text
+metadata but are not key path components.
+
+Identity digest/key use the documented stable identity subset and exclude
+`persist_mode` and `computed_at`; both remain wire metadata. Wire and identity
+digests are deterministic canonical JSON hashes. No legacy/default/ANY/alias
+fallback or store I/O is part of N1.

src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py

@@ -0,0 +1,209 @@
+"""Pure clean-slate qpk-vnext/result/v2 contract.
+
+Identity subset (and only this subset) is namespace, schema_version, domain,
+profile, timing, 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 never affect identity.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from datetime import datetime
+from types import MappingProxyType
+from typing import Any, Mapping
+
+MAX_SAFE_JSON_INTEGER = 2**53 - 1
+NAMESPACE = "qpk-vnext/result/v2"
+SCHEMA_VERSION = 2
+_WIRE_FIELDS = frozenset({
+    "namespace", "schema_version", "domain", "profile", "timing",
+    "identity_version", "persist_mode", "strategy_id", "run_id",
+    "param_set_id", "param_version", "source_revision", "computed_at",
+    "params", "identity_digest", "wire_digest",
+})
+_SAFE_SEGMENT = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,99}$")
+_TIMINGS = frozenset({"next_open", "next_close"})
+_MODES = frozenset({"durable", "ephemeral"})
+_MAX_DEPTH = 16
+_MAX_SEQUENCE = 128
+
+
+class ContractError(ValueError):
+    """Sanitized contract validation failure."""
+
+
+def _fail() -> None:
+    raise ContractError("invalid qpk-vnext contract")
+
+
+def _text(value: Any, *, bounded: bool = True) -> str:
+    if not isinstance(value, str):
+        _fail()
+    try:
+        value.encode("utf-8")
+    except UnicodeEncodeError:
+        _fail()
+    if any(0xD800 <= ord(ch) <= 0xDFFF for ch in value):
+        _fail()
+    if bounded and (not value or len(value) > 256):
+        _fail()
+    return value
+
+
+def _segment(value: Any) -> str:
+    value = _text(value)
+    if not _SAFE_SEGMENT.fullmatch(value) or value in {".", ".."}:
+        _fail()
+    return value
+
+
+def _integer(value: Any) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= MAX_SAFE_JSON_INTEGER:
+        _fail()
+    return value
+
+
+def _timestamp(value: Any) -> str:
+    value = _text(value)
+    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z", value):
+        _fail()
+    try:
+        datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ")
+    except ValueError:
+        _fail()
+    return value
+
+
+def _freeze(value: Any, depth: int, *, wire: bool = False) -> Any:
+    if depth > _MAX_DEPTH:
+        _fail()
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, int):
+        if abs(value) > MAX_SAFE_JSON_INTEGER:
+            _fail()
+        return value
+    if isinstance(value, str):
+        return _text(value)
+    if isinstance(value, float) or value is None:
+        _fail()
+    if isinstance(value, tuple) or (wire and isinstance(value, list)):
+        if len(value) > _MAX_SEQUENCE:
+            _fail()
+        return tuple(_freeze(item, depth + 1, wire=wire) for item in value)
+    if isinstance(value, Mapping):
+        _fail()
+    _fail()
+
+
+def _params(value: Any, *, wire: bool = False) -> MappingProxyType:
+    if not isinstance(value, Mapping):
+        _fail()
+    if len(value) > _MAX_SEQUENCE:
+        _fail()
+    result: dict[str, Any] = {}
+    for key, item in value.items():
+        key = _text(key)
+        result[key] = _freeze(item, 1, wire=wire)
+    return MappingProxyType(dict(sorted(result.items())))
+
+
+def _thaw(value: Any) -> Any:
+    if isinstance(value, tuple):
+        return [_thaw(item) for item in value]
+    if isinstance(value, Mapping):
+        return {key: _thaw(item) for key, item in value.items()}
+    return value
+
+
+def _canonical(value: Mapping[str, Any]) -> bytes:
+    try:
+        return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
+    except (TypeError, ValueError, UnicodeEncodeError):
+        _fail()
+
+
+def _digest(value: Mapping[str, Any]) -> str:
+    return hashlib.sha256(_canonical(value)).hexdigest()
+
+
+@dataclass(frozen=True)
+class ResultContract:
+    domain: str
+    profile: str
+    timing: str
+    identity_version: int
+    persist_mode: str
+    strategy_id: str
+    run_id: str
+    param_set_id: str
+    param_version: int
+    source_revision: str
+    computed_at: str
+    params: Mapping[str, Any]
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "domain", _segment(self.domain))
+        profile = _segment(self.profile)
+        if profile != profile.upper():
+            _fail()
+        object.__setattr__(self, "profile", profile)
+        if self.timing not in _TIMINGS or self.persist_mode not in _MODES:
+            _fail()
+        object.__setattr__(self, "identity_version", _integer(self.identity_version))
+        object.__setattr__(self, "param_version", _integer(self.param_version))
+        for name in ("strategy_id", "run_id"):
+            object.__setattr__(self, name, _segment(getattr(self, name)))
+        for name in ("param_set_id", "source_revision"):
+            object.__setattr__(self, name, _text(getattr(self, name)))
+        object.__setattr__(self, "computed_at", _timestamp(self.computed_at))
+        object.__setattr__(self, "params", _params(self.params))
+
+    def _identity(self) -> dict[str, Any]:
+        return {
+            "namespace": NAMESPACE, "schema_version": SCHEMA_VERSION,
+            "domain": self.domain, "profile": self.profile, "timing": self.timing,
+            "identity_version": self.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(self.params),
+        }
+
+    def to_wire(self) -> dict[str, Any]:
+        payload = {**self._identity(), "persist_mode": self.persist_mode, "computed_at": self.computed_at}
+        payload["identity_digest"] = _digest(self._identity())
+        payload["wire_digest"] = _digest(payload)
+        return payload
+
+    @property
+    def key(self) -> str:
+        return "/".join((NAMESPACE, self.domain, self.profile, self.strategy_id,
+                          self.run_id, self.timing, f"i{self.identity_version}",
+                          f"p{self.param_version}", f"{_digest(self._identity())}.json"))
+
+
+def decode_wire(data: Any) -> ResultContract:
+    try:
+        if not isinstance(data, Mapping) or set(data) != _WIRE_FIELDS:
+            _fail()
+        if data["namespace"] != NAMESPACE or data["schema_version"] != SCHEMA_VERSION:
+            _fail()
+        params = _params(data["params"], wire=True)
+        item = ResultContract(domain=data["domain"], profile=data["profile"], timing=data["timing"],
+                              identity_version=data["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"],
+                              source_revision=data["source_revision"], computed_at=data["computed_at"], params=params)
+        if data["identity_digest"] != _digest(item._identity()):
+            _fail()
+        expected = item.to_wire()
+        if data != expected:
+            _fail()
+        return item
+    except ContractError:
+        raise
+    except Exception:
+        raise ContractError("invalid qpk-vnext contract") from None

tests/test_qpk_vnext_n1.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from dataclasses import replace
+
+from quant_platform_kit.strategy_lifecycle.qpk_vnext_n1 import (
+    MAX_SAFE_JSON_INTEGER, ContractError, ResultContract, decode_wire,
+)
+
+
+def result(**overrides):
+    values = dict(domain="us_equity", profile="SOXL", timing="next_open", identity_version=1,
+                  persist_mode="durable", strategy_id="strategy", run_id="run-1",
+                  param_set_id="baseline", param_version=1, source_revision="rev1",
+                  computed_at="2026-07-15T00:00:00.000000Z", params={"x": 1, "seq": (True, ("a", 2))})
+    values.update(overrides)
+    return ResultContract(**values)
+
+
+class QpkVNextN1Tests(unittest.TestCase):
+    def test_profiles_roundtrip_and_identity_metadata_invariance(self):
+        for profile in ("SOXL", "TQQQ", "UPRO"):
+            item = result(profile=profile)
+            self.assertEqual(decode_wire(item.to_wire()), item)
+        changed = replace(result(), persist_mode="ephemeral", computed_at="2026-07-16T00:00:00.000000Z")
+        self.assertEqual(result().key, changed.key)
+        self.assertNotEqual(result().to_wire()["wire_digest"], changed.to_wire()["wire_digest"])
+
+    def test_identity_changes_and_permutation(self):
+        first = result(params={"b": 2, "a": 1})
+        second = result(params={"a": 1, "b": 2})
+        self.assertEqual(first.to_wire(), second.to_wire())
+        for field, value in (("run_id", "run-2"), ("timing", "next_close"), ("source_revision", "rev2"), ("param_version", 2)):
+            self.assertNotEqual(first.key, replace(first, **{field: value}).key)
+
+    def test_safe_ints_bool_and_decimal_policy(self):
+        self.assertEqual(result(params={"n": MAX_SAFE_JSON_INTEGER, "flag": True}).params["n"], MAX_SAFE_JSON_INTEGER)
+        for value in (MAX_SAFE_JSON_INTEGER + 1, -(MAX_SAFE_JSON_INTEGER + 1)):
+            with self.assertRaises(ContractError):
+                result(params={"n": value})
+        self.assertEqual(decode_wire(result(params={"decimal": "1.2500", "scaled": 12500}).to_wire()).params["scaled"], 12500)
+
+    def test_sequence_freeze_and_nested_mapping_rejection(self):
+        item = result()
+        self.assertEqual(decode_wire(item.to_wire()), item)
+        with self.assertRaises(AttributeError):
+            item.params["seq"].append(1)
+        with self.assertRaises(ContractError):
+            result(params={"nested": {"x": 1}})
+        wire = item.to_wire()
+        wire["params"] = {"nested": {"x": 1}}
+        with self.assertRaises(ContractError):
+            decode_wire(wire)
+
+    def test_strict_profile_timestamp_wire_and_values(self):
+        for profile in ("soxl", "A/B", "..", "X" * 101, ""):
+            with self.assertRaises(ContractError):
+                result(profile=profile)
+        for timestamp in ("2026-07-15T00:00:00Z", "2026-07-15T00:00:00.1Z", "2026-07-15T00:00:00.000000+00:00", "2026-02-30T00:00:00.000000Z"):
+            with self.assertRaises(ContractError):
+                result(computed_at=timestamp)
+        for value in (1.0, -0.0, float("nan"), float("inf"), float("-inf")):
+            with self.assertRaises(ContractError):
+                result(params={"v": value})
+
+    def test_unicode_safety_and_exact_wire_shape(self):
+        for kwargs in ({"strategy_id": "bad\ud800"}, {"params": {"bad\ud800": 1}}, {"params": {"v": "bad\ud800"}}):
+            with self.assertRaises(ContractError):
+                result(**kwargs)
+        wire = result().to_wire()
+        for mutation in ({"unknown": 1}, {"namespace": "legacy"}, {"computed_at": None}, {"timing": None}):
+            candidate = copy.deepcopy(wire)
+            candidate.update(mutation)
+            with self.assertRaises(ContractError):
+                decode_wire(candidate)
+
+    def test_key_bearing_ids_are_safe_segments(self):
+        for value in ("Mixed_ID-1", "A.1"):
+            item = result(strategy_id=value, run_id=value)
+            self.assertEqual(decode_wire(item.to_wire()), item)
+            self.assertEqual(len(item.key.split("/")), 11)
+        for field in ("strategy_id", "run_id"):
+            for value in ("", "/bad", "..", ".", "bad/control\n", "-bad", "X" * 101):
+                with self.assertRaises(ContractError):
+                    result(**{field: value})
+                wire = result().to_wire()
+                wire[field] = value
+                with self.assertRaises(ContractError):
+                    decode_wire(wire)
+
+
+if __name__ == "__main__":
+    unittest.main()