From ac20c8d92661051d05498470b10864152a13aa21 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:18:02 +0800 Subject: [PATCH] feat: verify portable decision price-series artifacts Co-Authored-By: Codex --- docs/decision-data-binding.zh-CN.md | 13 + src/quant_platform_kit/data/__init__.py | 18 + .../data/decision_price_series_artifact.py | 315 ++++++++++++++++++ tests/test_decision_price_series_artifact.py | 170 ++++++++++ 4 files changed, 516 insertions(+) create mode 100644 src/quant_platform_kit/data/decision_price_series_artifact.py create mode 100644 tests/test_decision_price_series_artifact.py diff --git a/docs/decision-data-binding.zh-CN.md b/docs/decision-data-binding.zh-CN.md index de0a1cc..916cadc 100644 --- a/docs/decision-data-binding.zh-CN.md +++ b/docs/decision-data-binding.zh-CN.md @@ -18,3 +18,16 @@ P1 数据采集 / 多源校验 → 冻结数据工件 → 策略决策 - `artifact_required`：策略历史输入必须匹配冻结工件和绑定哈希；缺失或非 `VERIFIED` 时，运行时应禁止新增风险。 `DecisionDataArtifactPort` 仅加载已验证的历史决策数据；`ExecutionQuotePort` 仅提供短时执行报价。原有 `MarketDataPort` 在迁移期间保留兼容，新的策略代码不应再把它同时用于历史决策和实时下单保护。 + +## 可移植日线投影（v1） + +P1 的原始根目录可以随策略而不同：例如 TQQQ 是 OHLCV，SOXL 的部分候选只需要复权收盘价。运行平台不能猜测这些私有格式。因此，准备让运行时消费的 P1 根目录必须额外包含： + +- `decision-price-series.json`：`qpk.decision_price_series_artifact.v1`；按标的提供日线 `as_of`、`close` 和可空的 `volume`。 +- `manifest.json`：继续使用 `research_input_manifest.v1`，并把上面文件的大小和 SHA-256 作为成员写入。 + +运行端先用公开绑定中的 `artifact_sha256` 校验 **规范化** `manifest.json`，再校验 `decision-price-series.json` 的成员哈希，最后比对 `strategy_scope`、最后交易日、复权口径和来源 ID。任何一步不匹配都不能返回历史序列。 + +这里没有把 `binding_sha256` 写进投影，以避免“绑定包含清单摘要、清单又包含投影”的循环哈希。运行时会在读回投影后重新计算并验证绑定身份。 + +存储地址只能由部署环境的私有解析器提供，不能出现在 `RuntimeTarget`、控制台页面或执行回报中。P1 仍须先完成自身多源校验；这个投影只是经过 P1 验证后给平台使用的通用出口，不替代 P1、P3 或 P4--P6 的晋升门槛。 diff --git a/src/quant_platform_kit/data/__init__.py b/src/quant_platform_kit/data/__init__.py index 7047ec3..f0584ad 100644 --- a/src/quant_platform_kit/data/__init__.py +++ b/src/quant_platform_kit/data/__init__.py @@ -42,6 +42,16 @@ DECISION_DATA_MODE_LEGACY_RUNTIME_FETCH, DecisionDataBinding, ) +from quant_platform_kit.data.decision_price_series_artifact import ( + DECISION_PRICE_SERIES_ARTIFACT_SCHEMA_VERSION, + DECISION_PRICE_SERIES_MEMBER_PATH, + InvalidDecisionDataArtifact, + canonical_decision_price_series_artifact_bytes, + price_series_from_decision_price_series_artifact, + read_decision_price_series_artifact_json, + validate_decision_price_series_artifact, + verify_decision_price_series_artifact_members, +) __all__ = [ "DataVersion", @@ -63,15 +73,23 @@ "SOURCE_OBSERVATION_UNAVAILABLE", "DailyBar", "DecisionDataBinding", + "DECISION_PRICE_SERIES_ARTIFACT_SCHEMA_VERSION", + "DECISION_PRICE_SERIES_MEMBER_PATH", "DailyBarSourceObservation", "DailyBarSourceSnapshot", "MultiSourceDailyBarAssurance", "MultiSourceDailyBarPolicy", + "InvalidDecisionDataArtifact", "assess_multisource_daily_bars", + "canonical_decision_price_series_artifact_bytes", "build_artifact_record", "latest_version", + "price_series_from_decision_price_series_artifact", + "read_decision_price_series_artifact_json", "resolve_version", "semver_version", + "validate_decision_price_series_artifact", + "verify_decision_price_series_artifact_members", "write_artifact_manifest", "write_data_release", ] diff --git a/src/quant_platform_kit/data/decision_price_series_artifact.py b/src/quant_platform_kit/data/decision_price_series_artifact.py new file mode 100644 index 0000000..6e083e5 --- /dev/null +++ b/src/quant_platform_kit/data/decision_price_series_artifact.py @@ -0,0 +1,315 @@ +"""Portable, verified daily decision-data projections. + +This contract intentionally carries historical decision inputs only. It does +not describe a storage location, a broker account, or a short-lived execution +quote. Pipeline-specific P1 validators remain responsible for proving their +native input root; this module verifies the small portable projection that a +runtime may consume after that proof has been published immutably. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import UTC, date, datetime, time +from hashlib import sha256 +import json +from math import isfinite +import re +from typing import Any + +from quant_platform_kit.common.models import PricePoint, PriceSeries + +from .decision_data_binding import ( + DECISION_DATA_ASSURANCE_VERIFIED, + DECISION_DATA_MODE_ARTIFACT_OPTIONAL, + DECISION_DATA_MODE_ARTIFACT_REQUIRED, + DecisionDataBinding, +) +from .research_input import ( + canonical_research_input_manifest_bytes, + read_research_input_manifest_json, + research_input_manifest_sha256, +) + + +DECISION_PRICE_SERIES_ARTIFACT_SCHEMA_VERSION = "qpk.decision_price_series_artifact.v1" +DECISION_PRICE_SERIES_MEMBER_PATH = "decision-price-series.json" + +_SYMBOL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") +_CURRENCY_RE = re.compile(r"^[A-Z]{2,8}$") + + +class InvalidDecisionDataArtifact(ValueError): + """Raised when a portable decision-data projection fails closed.""" + + +def _invalid() -> None: + raise InvalidDecisionDataArtifact("invalid decision price-series artifact") + + +def _canonical(value: object) -> bytes: + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + _invalid() + result[key] = value + return result + + +def _reject_nonfinite_constant(_: str) -> None: + _invalid() + + +def _require_exact_mapping(value: object, keys: frozenset[str]) -> dict[str, object]: + if not isinstance(value, Mapping) or set(value) != keys or any( + not isinstance(key, str) for key in value + ): + _invalid() + return dict(value) + + +def _require_date(value: object) -> str: + if not isinstance(value, str): + _invalid() + try: + parsed = date.fromisoformat(value) + except ValueError: + _invalid() + if parsed.isoformat() != value: + _invalid() + return value + + +def _require_identifier(value: object) -> str: + if not isinstance(value, str): + _invalid() + text = value.strip() + if not text or text != value: + _invalid() + return text + + +def _require_number(value: object, *, positive: bool) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + _invalid() + number = float(value) + if not isfinite(number) or (number <= 0 if positive else number < 0): + _invalid() + return number + + +def _require_source_ids(value: object) -> tuple[str, ...]: + if not isinstance(value, list) or not value: + _invalid() + source_ids = tuple(_require_identifier(item) for item in value) + if len(source_ids) != len(set(source_ids)): + _invalid() + return source_ids + + +def validate_decision_price_series_artifact(value: object) -> dict[str, object]: + """Validate one provider-neutral, daily historical price projection.""" + + try: + artifact = _require_exact_mapping( + value, + frozenset( + { + "schema_version", + "strategy_scope", + "as_of", + "adjustment_basis", + "source_ids", + "series", + } + ), + ) + if artifact["schema_version"] != DECISION_PRICE_SERIES_ARTIFACT_SCHEMA_VERSION: + _invalid() + strategy_scope = _require_identifier(artifact["strategy_scope"]) + as_of = _require_date(artifact["as_of"]) + adjustment_basis = _require_identifier(artifact["adjustment_basis"]) + source_ids = _require_source_ids(artifact["source_ids"]) + raw_series = artifact["series"] + if not isinstance(raw_series, Mapping) or not raw_series: + _invalid() + + normalized_series: dict[str, dict[str, object]] = {} + for raw_symbol, raw_payload in raw_series.items(): + if not isinstance(raw_symbol, str) or not _SYMBOL_RE.fullmatch(raw_symbol): + _invalid() + symbol = raw_symbol.upper() + if symbol != raw_symbol or symbol in normalized_series: + _invalid() + payload = _require_exact_mapping(raw_payload, frozenset({"currency", "points"})) + currency = _require_identifier(payload["currency"]) + if not _CURRENCY_RE.fullmatch(currency): + _invalid() + raw_points = payload["points"] + if not isinstance(raw_points, list) or not raw_points: + _invalid() + + points: list[dict[str, object]] = [] + prior_session: str | None = None + for raw_point in raw_points: + point = _require_exact_mapping(raw_point, frozenset({"as_of", "close", "volume"})) + session = _require_date(point["as_of"]) + if prior_session is not None and session <= prior_session: + _invalid() + prior_session = session + if session > as_of: + _invalid() + volume_raw = point["volume"] + volume = None if volume_raw is None else _require_number(volume_raw, positive=False) + points.append( + { + "as_of": session, + "close": _require_number(point["close"], positive=True), + "volume": volume, + } + ) + if prior_session != as_of: + _invalid() + normalized_series[symbol] = {"currency": currency, "points": points} + + return { + "schema_version": DECISION_PRICE_SERIES_ARTIFACT_SCHEMA_VERSION, + "strategy_scope": strategy_scope, + "as_of": as_of, + "adjustment_basis": adjustment_basis, + "source_ids": list(source_ids), + "series": normalized_series, + } + except (InvalidDecisionDataArtifact, TypeError, ValueError, KeyError): + raise InvalidDecisionDataArtifact("invalid decision price-series artifact") from None + + +def canonical_decision_price_series_artifact_bytes(value: object) -> bytes: + """Return canonical bytes after strict validation.""" + + return _canonical(validate_decision_price_series_artifact(value)) + + +def read_decision_price_series_artifact_json(payload: bytes | str) -> dict[str, object]: + """Strictly parse a portable projection without accepting duplicate keys.""" + + try: + if isinstance(payload, bytes): + payload = payload.decode("utf-8") + if not isinstance(payload, str): + _invalid() + parsed = json.loads( + payload, + object_pairs_hook=_reject_duplicate_keys, + parse_constant=_reject_nonfinite_constant, + ) + return validate_decision_price_series_artifact(parsed) + except (InvalidDecisionDataArtifact, UnicodeDecodeError, TypeError, ValueError, json.JSONDecodeError): + raise InvalidDecisionDataArtifact("invalid decision price-series artifact") from None + + +def _require_verified_artifact_binding(binding: DecisionDataBinding) -> None: + if ( + binding.mode + not in { + DECISION_DATA_MODE_ARTIFACT_OPTIONAL, + DECISION_DATA_MODE_ARTIFACT_REQUIRED, + } + or binding.assurance_status != DECISION_DATA_ASSURANCE_VERIFIED + ): + raise InvalidDecisionDataArtifact("decision data binding is not verified for artifact use") + + +def price_series_from_decision_price_series_artifact( + artifact: object, + *, + binding: DecisionDataBinding, +) -> dict[str, PriceSeries]: + """Translate a verified projection only when its public identity matches.""" + + _require_verified_artifact_binding(binding) + normalized = validate_decision_price_series_artifact(artifact) + if ( + normalized["strategy_scope"] != binding.strategy_scope + or normalized["as_of"] != binding.as_of + or normalized["adjustment_basis"] != binding.adjustment_basis + or tuple(normalized["source_ids"]) != binding.source_ids + ): + raise InvalidDecisionDataArtifact("decision price-series artifact does not match binding") + + result: dict[str, PriceSeries] = {} + for symbol, raw_payload in normalized["series"].items(): + payload = _require_exact_mapping(raw_payload, frozenset({"currency", "points"})) + points = tuple( + PricePoint( + as_of=datetime.combine(date.fromisoformat(point["as_of"]), time.min, tzinfo=UTC), + close=float(point["close"]), + volume=(None if point["volume"] is None else float(point["volume"])), + ) + for point in payload["points"] + ) + result[symbol] = PriceSeries(symbol=symbol, currency=str(payload["currency"]), points=points) + return result + + +def verify_decision_price_series_artifact_members( + *, + binding: DecisionDataBinding, + manifest_bytes: bytes, + decision_price_series_bytes: bytes, +) -> dict[str, PriceSeries]: + """Verify immutable manifest/member bytes, then return safe price series. + + A transport adapter resolves the private root. This function deliberately + receives only bytes, so storage paths and credentials cannot enter the + public runtime target or its execution report. + """ + + try: + _require_verified_artifact_binding(binding) + manifest = read_research_input_manifest_json(manifest_bytes) + if manifest_bytes != canonical_research_input_manifest_bytes(manifest): + _invalid() + if research_input_manifest_sha256(manifest) != binding.artifact_sha256: + _invalid() + members = {str(member["path"]): member for member in manifest["members"]} + member = members.get(DECISION_PRICE_SERIES_MEMBER_PATH) + if not isinstance(member, Mapping): + _invalid() + if ( + member.get("size_bytes") != len(decision_price_series_bytes) + or member.get("sha256") != sha256(decision_price_series_bytes).hexdigest() + ): + _invalid() + if decision_price_series_bytes != canonical_decision_price_series_artifact_bytes( + read_decision_price_series_artifact_json(decision_price_series_bytes) + ): + _invalid() + return price_series_from_decision_price_series_artifact( + read_decision_price_series_artifact_json(decision_price_series_bytes), + binding=binding, + ) + except (InvalidDecisionDataArtifact, TypeError, ValueError, KeyError): + raise InvalidDecisionDataArtifact("invalid decision price-series artifact") from None + + +__all__ = [ + "DECISION_PRICE_SERIES_ARTIFACT_SCHEMA_VERSION", + "DECISION_PRICE_SERIES_MEMBER_PATH", + "InvalidDecisionDataArtifact", + "canonical_decision_price_series_artifact_bytes", + "price_series_from_decision_price_series_artifact", + "read_decision_price_series_artifact_json", + "validate_decision_price_series_artifact", + "verify_decision_price_series_artifact_members", +] diff --git a/tests/test_decision_price_series_artifact.py b/tests/test_decision_price_series_artifact.py new file mode 100644 index 0000000..733b2ec --- /dev/null +++ b/tests/test_decision_price_series_artifact.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +from hashlib import sha256 + +import pytest + +from quant_platform_kit.data import ( + DECISION_DATA_ASSURANCE_VERIFIED, + DECISION_DATA_MODE_ARTIFACT_REQUIRED, + DECISION_PRICE_SERIES_MEMBER_PATH, + DecisionDataBinding, + InvalidDecisionDataArtifact, + canonical_decision_price_series_artifact_bytes, + read_decision_price_series_artifact_json, + verify_decision_price_series_artifact_members, +) +from quant_platform_kit.data.research_input import ( + canonical_research_input_manifest_bytes, + read_research_input_manifest_json, + research_input_manifest_sha256, +) + + +def _artifact() -> dict[str, object]: + return { + "schema_version": "qpk.decision_price_series_artifact.v1", + "strategy_scope": "tqqq_growth_income", + "as_of": "2026-08-28", + "adjustment_basis": "split_adjusted", + "source_ids": ["twelve_data_daily", "yahoo_finance_daily"], + "series": { + "QQQ": { + "currency": "USD", + "points": [ + {"as_of": "2026-08-27", "close": 100.0, "volume": 123.0}, + {"as_of": "2026-08-28", "close": 101.5, "volume": None}, + ], + } + }, + } + + +def _manifest(projection: bytes) -> bytes: + return canonical_research_input_manifest_bytes( + { + "schema_version": "research_input_manifest.v1", + "manifest_id": "decision-price-series-test", + "research_input_contract_id": "tqqq_daily_decision_input.v1", + "domain": "us_equity", + "profile": "tqqq_growth_income", + "artifact_type": "immutable_assured_daily_decision_price_series", + "observed_at": "2026-08-28T20:00:00Z", + "effective_at": "2026-08-28T20:00:00Z", + "as_of": "2026-08-28T20:00:00Z", + "producer": { + "repository": "QuantStrategyLab/UsEquitySnapshotPipelines", + "commit_sha": "a" * 40, + "tree_sha": "b" * 40, + "tool": "daily_decision_projection", + "tool_version": "v1", + }, + "calendar": { + "calendar_id": "XNYS", + "timezone": "America/New_York", + "session_date": "2026-08-28", + "source": "exchange_calendars", + "source_revision": "v1", + }, + "adjustment": { + "policy": "split_adjusted", + "source": "two_source_assurance", + "source_revision": "v1", + }, + "sources": [ + { + "source_id": "twelve_data_daily:QQQ", + "revision": "v1", + "observed_at": "2026-08-28T20:00:00Z", + "content_sha256": "c" * 64, + }, + { + "source_id": "yahoo_finance_daily:QQQ", + "revision": "v1", + "observed_at": "2026-08-28T20:00:00Z", + "content_sha256": "d" * 64, + }, + ], + "members": [ + { + "path": DECISION_PRICE_SERIES_MEMBER_PATH, + "media_type": "application/json", + "size_bytes": len(projection), + "sha256": sha256(projection).hexdigest(), + } + ], + } + ) + + +def _binding(manifest: bytes) -> DecisionDataBinding: + return DecisionDataBinding( + binding_id="tqqq-daily-decision-data-v1", + strategy_scope="tqqq_growth_income", + mode=DECISION_DATA_MODE_ARTIFACT_REQUIRED, + source_ids=("twelve_data_daily", "yahoo_finance_daily"), + as_of="2026-08-28", + adjustment_basis="split_adjusted", + artifact_sha256=research_input_manifest_sha256(read_research_input_manifest_json(manifest)), + assurance_status=DECISION_DATA_ASSURANCE_VERIFIED, + ) + + +def test_verified_projection_requires_matching_canonical_manifest_and_member() -> None: + projection = canonical_decision_price_series_artifact_bytes(_artifact()) + manifest = _manifest(projection) + + series = verify_decision_price_series_artifact_members( + binding=_binding(manifest), + manifest_bytes=manifest, + decision_price_series_bytes=projection, + ) + + assert list(series) == ["QQQ"] + assert series["QQQ"].currency == "USD" + assert series["QQQ"].latest.close == 101.5 + assert series["QQQ"].latest.volume is None + assert series["QQQ"].latest.as_of.isoformat() == "2026-08-28T00:00:00+00:00" + + +def test_projection_rejects_binding_identity_or_member_integrity_mismatch() -> None: + projection = canonical_decision_price_series_artifact_bytes(_artifact()) + manifest = _manifest(projection) + binding = _binding(manifest) + + with pytest.raises(InvalidDecisionDataArtifact): + verify_decision_price_series_artifact_members( + binding=binding, + manifest_bytes=manifest, + decision_price_series_bytes=projection + b" ", + ) + + with pytest.raises(InvalidDecisionDataArtifact): + verify_decision_price_series_artifact_members( + binding=binding, + manifest_bytes=manifest + b"\n", + decision_price_series_bytes=projection, + ) + + mismatched = _artifact() + mismatched["source_ids"] = ["twelve_data_daily"] + with pytest.raises(InvalidDecisionDataArtifact): + verify_decision_price_series_artifact_members( + binding=binding, + manifest_bytes=manifest, + decision_price_series_bytes=canonical_decision_price_series_artifact_bytes(mismatched), + ) + + +def test_projection_rejects_duplicate_json_keys_and_incomplete_daily_series() -> None: + with pytest.raises(InvalidDecisionDataArtifact): + read_decision_price_series_artifact_json( + b'{"schema_version":"qpk.decision_price_series_artifact.v1","schema_version":"x"}' + ) + + incomplete = _artifact() + incomplete["series"] = { + "QQQ": {"currency": "USD", "points": [{"as_of": "2026-08-27", "close": 1.0, "volume": None}]} + } + with pytest.raises(InvalidDecisionDataArtifact): + canonical_decision_price_series_artifact_bytes(incomplete)