From ef1933dfed05f56a5db7658fd828dca5899a1f77 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:52:03 +0800 Subject: [PATCH 1/2] feat: add converged qpk vnext n1 contract Co-Authored-By: Codex --- docs/qpk_vnext_n1_param_grammar.md | 21 ++ .../strategy_lifecycle/qpk_vnext_n1.py | 207 ++++++++++++++++++ tests/test_qpk_vnext_n1.py | 80 +++++++ 3 files changed, 308 insertions(+) create mode 100644 docs/qpk_vnext_n1_param_grammar.md create mode 100644 src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py create mode 100644 tests/test_qpk_vnext_n1.py diff --git a/docs/qpk_vnext_n1_param_grammar.md b/docs/qpk_vnext_n1_param_grammar.md new file mode 100644 index 0000000..a4be8f5 --- /dev/null +++ b/docs/qpk_vnext_n1_param_grammar.md @@ -0,0 +1,21 @@ +# QPK vNext N1 Params Grammar + +Approved clean-slate source of truth for `qpk-vnext/result/v2`: + +```text +Params := JSON object / safe Mapping (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. + +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. diff --git a/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py new file mode 100644 index 0000000..b008de7 --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py @@ -0,0 +1,207 @@ +"""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", "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 diff --git a/tests/test_qpk_vnext_n1.py b/tests/test_qpk_vnext_n1.py new file mode 100644 index 0000000..1ad6fb0 --- /dev/null +++ b/tests/test_qpk_vnext_n1.py @@ -0,0 +1,80 @@ +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) + + +if __name__ == "__main__": + unittest.main() From 001217b75556f4188fd92f763605da4824d7dff8 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:57:49 +0800 Subject: [PATCH 2/2] fix: constrain key-bearing result ids Co-Authored-By: Codex --- docs/qpk_vnext_n1_param_grammar.md | 6 ++++++ .../strategy_lifecycle/qpk_vnext_n1.py | 4 +++- tests/test_qpk_vnext_n1.py | 14 ++++++++++++++ 3 files changed, 23 insertions(+), 1 deletion(-) diff --git a/docs/qpk_vnext_n1_param_grammar.md b/docs/qpk_vnext_n1_param_grammar.md index a4be8f5..9288c5e 100644 --- a/docs/qpk_vnext_n1_param_grammar.md +++ b/docs/qpk_vnext_n1_param_grammar.md @@ -15,6 +15,12 @@ surrogates, and excessive recursion/size fail closed with sanitized contract err 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 diff --git a/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py index b008de7..cbf9153 100644 --- a/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py +++ b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n1.py @@ -155,7 +155,9 @@ def __post_init__(self) -> None: _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", "param_set_id", "source_revision"): + 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)) diff --git a/tests/test_qpk_vnext_n1.py b/tests/test_qpk_vnext_n1.py index 1ad6fb0..e704c5e 100644 --- a/tests/test_qpk_vnext_n1.py +++ b/tests/test_qpk_vnext_n1.py @@ -75,6 +75,20 @@ def test_unicode_safety_and_exact_wire_shape(self): 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()