From 4dc87d08c32d961272d8a5915e8aaf4f682660fb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:34:02 +0800 Subject: [PATCH 1/2] feat: add converged clean-slate vnext contract Co-Authored-By: Codex --- .../strategy_lifecycle/vnext_n1_contract.py | 150 ++++++++++++++++++ tests/test_vnext_n1_converged.py | 79 +++++++++ 2 files changed, 229 insertions(+) create mode 100644 src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py create mode 100644 tests/test_vnext_n1_converged.py diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py new file mode 100644 index 0000000..0a032ca --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py @@ -0,0 +1,150 @@ +"""Clean-slate qpk-vnext/result/v2 JSON contract; no legacy or I/O fallback. + +Identity fields are namespace/schema, 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-only +metadata. Higher layers own profile allowlists and persistence side effects. +""" + +from __future__ import annotations + +import hashlib +import json +import math +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 = "qpk.vnext.result.v2" +_MAX_SEGMENT = 100 +_MAX_DEPTH = 16 +_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$") +_FIELDS = frozenset({"namespace", "schema_version", "domain", "canonical_profile", "execution_timing", "result_identity_version", "persist_mode", "strategy_id", "run_id", "param_set_id", "param_version", "source_revision", "computed_at", "params", "identity_digest", "wire_digest"}) + + +class VNextContractError(ValueError): + def __init__(self) -> None: + super().__init__("invalid qpk-vnext result contract") + + +def _segment(value: Any, *, uppercase: bool = False) -> str: + if not isinstance(value, str) or not value or len(value) > _MAX_SEGMENT or value != value.strip() or value != value.strip("-._"): + raise VNextContractError() + if value in {".", ".."} or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for c in value): + raise VNextContractError() + if uppercase and value != value.upper(): + raise VNextContractError() + return value + + +def _integer(value: Any) -> int: + if not isinstance(value, int) or isinstance(value, bool) or value < 1 or value > MAX_SAFE_JSON_INTEGER: + raise VNextContractError() + return value + + +def _freeze(value: Any, depth: int = 0, *, wire: bool = False) -> Any: + if depth > _MAX_DEPTH or 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) or (wire and isinstance(value, list)): + return tuple(_freeze(item, depth + 1, wire=wire) for item in value) + raise VNextContractError() + + +def _params(value: Any, *, wire: bool = False) -> MappingProxyType: + if not isinstance(value, Mapping) or any(not isinstance(key, str) or not key for key in value): + raise VNextContractError() + return MappingProxyType({key: _freeze(value[key], wire=wire) for key in sorted(value)}) + + +def _thaw(value: Any) -> Any: + if isinstance(value, tuple): + return [_thaw(item) for item in value] + return value + + +def _json(value: Mapping[str, Any]) -> bytes: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + + +def _hash(value: Mapping[str, Any]) -> str: + return hashlib.sha256(_json(value)).hexdigest() + + +def _timestamp(value: Any) -> str: + if not isinstance(value, str) or not _TIMESTAMP.fullmatch(value): + raise VNextContractError() + try: + datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ") + except ValueError as exc: + raise VNextContractError() from exc + return value + + +@dataclass(frozen=True) +class VNextResult: + domain: str + canonical_profile: str + execution_timing: str + result_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: + _segment(self.domain) + _segment(self.canonical_profile, uppercase=True) + if self.execution_timing not in {"next_open", "next_close"} or self.persist_mode not in {"durable", "ephemeral"}: + raise VNextContractError() + for value in (self.strategy_id, self.run_id, self.param_set_id, self.source_revision): + _segment(value) + object.__setattr__(self, "result_identity_version", _integer(self.result_identity_version)) + object.__setattr__(self, "param_version", _integer(self.param_version)) + 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, "domain": self.domain, "canonical_profile": self.canonical_profile, "execution_timing": self.execution_timing, "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": {key: _thaw(self.params[key]) for key in sorted(self.params)}} + + def _wire_payload(self) -> dict[str, Any]: + return {**self._identity(), "persist_mode": self.persist_mode, "computed_at": self.computed_at} + + def to_wire(self) -> dict[str, Any]: + payload = self._wire_payload() + return {**payload, "identity_digest": _hash(self._identity()), "wire_digest": _hash(payload)} + + @property + def key(self) -> str: + return f"{NAMESPACE}/{self.domain}/{self.canonical_profile}/{self.strategy_id}/{self.run_id}/{self.execution_timing}/i{self.result_identity_version}/p{self.param_version}/{_hash(self._identity())}.json" + + +def decode_wire(data: Mapping[str, Any]) -> VNextResult: + try: + if not isinstance(data, Mapping) or set(data) != _FIELDS or data["namespace"] != NAMESPACE or data["schema_version"] != SCHEMA: + raise VNextContractError() + params = _params(data["params"], wire=True) + item = VNextResult(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"], source_revision=data["source_revision"], computed_at=data["computed_at"], params=params) + if data["identity_digest"] != _hash(item._identity()) or data["wire_digest"] != _hash(item._wire_payload()) or item.to_wire() != dict(data): + raise VNextContractError() + return item + except VNextContractError: + raise + except Exception as exc: + raise VNextContractError() from exc diff --git a/tests/test_vnext_n1_converged.py b/tests/test_vnext_n1_converged.py new file mode 100644 index 0000000..8cafa18 --- /dev/null +++ b/tests/test_vnext_n1_converged.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import copy +import unittest +from dataclasses import replace + +from quant_platform_kit.strategy_lifecycle.vnext_n1_contract import ( + MAX_SAFE_JSON_INTEGER, + VNextContractError, + VNextResult, + decode_wire, +) + + +def result(**overrides): + values = dict( + domain="us_equity", canonical_profile="SOXL", execution_timing="next_open", + result_identity_version=1, persist_mode="durable", strategy_id="soxl_trend", + 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, "nested": (True, ("a", 2))}, + ) + values.update(overrides) + return VNextResult(**values) + + +class ConvergedN1Tests(unittest.TestCase): + def test_profiles_timing_and_round_trip(self): + for profile in ("SOXL", "TQQQ", "UPRO"): + item = result(canonical_profile=profile) + self.assertEqual(decode_wire(item.to_wire()), item) + + def test_identity_subset_and_metadata(self): + base = result() + changed = replace(base, persist_mode="ephemeral", computed_at="2026-07-16T00:00:00.000000Z") + self.assertEqual(base.key, changed.key) + self.assertEqual(base.to_wire()["identity_digest"], changed.to_wire()["identity_digest"]) + self.assertNotEqual(base.to_wire()["wire_digest"], changed.to_wire()["wire_digest"]) + for field, value in (("run_id", "run-2"), ("source_revision", "rev2"), ("execution_timing", "next_close"), ("param_version", 2)): + self.assertNotEqual(base.key, replace(base, **{field: value}).key) + + def test_safe_integer_all_positions_and_bool(self): + self.assertEqual(result(params={"n": MAX_SAFE_JSON_INTEGER}).params["n"], MAX_SAFE_JSON_INTEGER) + self.assertTrue(result(params={"flag": True}).params["flag"]) + for field in ("result_identity_version", "param_version"): + with self.assertRaises(VNextContractError): + result(**{field: MAX_SAFE_JSON_INTEGER + 1}) + with self.assertRaises(VNextContractError): + result(params={"n": MAX_SAFE_JSON_INTEGER + 1}) + + def test_deep_immutable_and_wire_lists(self): + item = result() + with self.assertRaises(AttributeError): + item.params["nested"].append(3) + with self.assertRaises(VNextContractError): + result(params={"x": [1]}) + wire = item.to_wire() + self.assertEqual(decode_wire(wire), item) + + def test_timestamp_and_profile_strictness(self): + 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", " 2026-07-15T00:00:00.000000Z"): + with self.assertRaises(VNextContractError): + result(computed_at=timestamp) + for profile in ("soxl", "A/B", "..", "X" * 101, ""): + with self.assertRaises(VNextContractError): + result(canonical_profile=profile) + + def test_wire_shape_namespace_and_adversarial_values(self): + wire = result().to_wire() + for mutation in ({"unknown": 1}, {"namespace": "legacy"}, {"computed_at": None}, {"params": {"x": [MAX_SAFE_JSON_INTEGER + 1]}}): + candidate = copy.deepcopy(wire) + candidate.update(mutation) + with self.assertRaises(VNextContractError): + decode_wire(candidate) + self.assertTrue(result().key.startswith("qpk-vnext/result/v2/")) + + +if __name__ == "__main__": + unittest.main() From c6e5ca559a4253b180c6c14f228211eccc9af992 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:39:49 +0800 Subject: [PATCH 2/2] fix: reject unicode surrogates and raw floats Co-Authored-By: Codex --- .../strategy_lifecycle/vnext_n1_contract.py | 25 ++++++++++++++++--- tests/test_vnext_n1_converged.py | 14 +++++++++++ 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py index 0a032ca..109899c 100644 --- a/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_n1_contract.py @@ -31,9 +31,18 @@ def __init__(self) -> None: super().__init__("invalid qpk-vnext result contract") +def _unicode(value: str) -> str: + try: + value.encode("utf-8") + except UnicodeEncodeError: + raise VNextContractError() from None + return value + + def _segment(value: Any, *, uppercase: bool = False) -> str: if not isinstance(value, str) or not value or len(value) > _MAX_SEGMENT or value != value.strip() or value != value.strip("-._"): raise VNextContractError() + _unicode(value) if value in {".", ".."} or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for c in value): raise VNextContractError() if uppercase and value != value.upper(): @@ -50,14 +59,17 @@ def _integer(value: Any) -> int: def _freeze(value: Any, depth: int = 0, *, wire: bool = False) -> Any: if depth > _MAX_DEPTH or value is None: raise VNextContractError() - if isinstance(value, (str, bool)): + if isinstance(value, str): + _unicode(value) + return value + if isinstance(value, 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, float): + raise VNextContractError() if isinstance(value, tuple) or (wire and isinstance(value, list)): return tuple(_freeze(item, depth + 1, wire=wire) for item in value) raise VNextContractError() @@ -66,6 +78,8 @@ def _freeze(value: Any, depth: int = 0, *, wire: bool = False) -> Any: def _params(value: Any, *, wire: bool = False) -> MappingProxyType: if not isinstance(value, Mapping) or any(not isinstance(key, str) or not key for key in value): raise VNextContractError() + for key in value: + _unicode(key) return MappingProxyType({key: _freeze(value[key], wire=wire) for key in sorted(value)}) @@ -76,7 +90,10 @@ def _thaw(value: Any) -> Any: def _json(value: Mapping[str, Any]) -> bytes: - return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (UnicodeEncodeError, ValueError): + raise VNextContractError() from None def _hash(value: Mapping[str, Any]) -> str: diff --git a/tests/test_vnext_n1_converged.py b/tests/test_vnext_n1_converged.py index 8cafa18..23efd23 100644 --- a/tests/test_vnext_n1_converged.py +++ b/tests/test_vnext_n1_converged.py @@ -74,6 +74,20 @@ def test_wire_shape_namespace_and_adversarial_values(self): decode_wire(candidate) self.assertTrue(result().key.startswith("qpk-vnext/result/v2/")) + def test_unicode_scalar_and_raw_float_policy(self): + for kwargs in ({"params": {"bad\ud800": 1}}, {"params": {"good": "bad\ud800"}}, {"strategy_id": "bad\ud800"}): + with self.assertRaises(VNextContractError): + result(**kwargs) + for value in (1.0, -0.0, float("nan"), float("inf"), float("-inf")): + with self.assertRaises(VNextContractError): + result(params={"value": value}) + decimal_like = result(params={"value": "1.2500", "scaled": 12500}) + self.assertEqual(decode_wire(decimal_like.to_wire()), decimal_like) + wire = result().to_wire() + wire["strategy_id"] = "bad\ud800" + with self.assertRaises(VNextContractError): + decode_wire(wire) + if __name__ == "__main__": unittest.main()