diff --git a/README.md b/README.md index 358e463..b911c5d 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,9 @@ session/trading-day rollup, or event-Bar threshold and exact source-range eviden verify the complete source snapshot, physical and logical partition hashes, PIT timestamps, context membership, ordering, L2 snapshot/delta replay, selection hash and a second post-read snapshot check. Legacy Curated manifests remain readable through `load_curated_snapshot` but fail -with `legacy-curated-not-m8-certified` at the certified factory. +with `legacy-curated-not-m8-certified` at the certified factory. `VerifiedFactorInput`是不可公开 +构造、不可`dataclasses.replace`且不可变的opaque结果；模块不暴露认证令牌或可直接调用的 +类工厂，只有上述两个完成全链路校验的loader会实例化它。 ## M2 data-lake guarantees diff --git a/src/quant_data_kit/_version.py b/src/quant_data_kit/_version.py index 0cc858e..746ade0 100644 --- a/src/quant_data_kit/_version.py +++ b/src/quant_data_kit/_version.py @@ -1,3 +1,3 @@ """Single authoritative package version used by builds and runtime imports.""" -__version__ = "0.8.0" +__version__ = "0.8.1" diff --git a/src/quant_data_kit/research_contracts_v2.py b/src/quant_data_kit/research_contracts_v2.py index 1b9ce1e..fdbe515 100644 --- a/src/quant_data_kit/research_contracts_v2.py +++ b/src/quant_data_kit/research_contracts_v2.py @@ -5,8 +5,8 @@ import hashlib import re from collections.abc import Mapping -from dataclasses import InitVar, dataclass, field -from typing import Any, Literal +from dataclasses import dataclass +from typing import Any, Literal, NoReturn import pyarrow as pa from pyarrow import ipc @@ -30,7 +30,6 @@ _CANONICAL_NONNEGATIVE = re.compile(r"^(?:0|[1-9][0-9]*)$") _CANONICAL_POSITIVE = re.compile(r"^[1-9][0-9]*$") _INT64_MAX = 2**63 - 1 -_VERIFIED_INPUT_FACTORY_TOKEN = object() def _required_text(value: str, field_name: str) -> str: @@ -381,103 +380,30 @@ def from_contract(cls, payload: Mapping[str, Any]) -> CuratedAggregation: ) -@dataclass(frozen=True, eq=False) class VerifiedFactorInput: - layer: Literal["curated", "normalized"] - source_snapshot_id: str - source_logical_sha256: str - selection_logical_sha256: str - event_schemas: tuple[EventSchemaRef, ...] - table: pa.Table = field(repr=False) - calendar_id: str = "" - session_policy_version: str = "" - market_context_snapshot_id: str = "" - market_context_logical_sha256: str = "" - lineage: tuple[LineageRef, ...] = () - aggregation: CuratedAggregation | None = None - schema_id: str = VERIFIED_FACTOR_INPUT_SCHEMA_ID - _factory_token: InitVar[object] = None - - def __post_init__(self, _factory_token: object) -> None: - if _factory_token is not _VERIFIED_INPUT_FACTORY_TOKEN: - raise ValidationError("VerifiedFactorInput can only be created by a certified factory") - if self.schema_id != VERIFIED_FACTOR_INPUT_SCHEMA_ID: - raise ValidationError("unsupported VerifiedFactorInput schema") - if self.layer not in {"curated", "normalized"}: - raise ValidationError("unsupported verified input layer") - _require_snapshot_id(self.source_snapshot_id, "source_snapshot_id") - _require_hash(self.source_logical_sha256, "source_logical_sha256") - _require_hash(self.selection_logical_sha256, "selection_logical_sha256") - _required_text(self.calendar_id, "calendar_id") - _required_text(self.session_policy_version, "session_policy_version") - _require_snapshot_id(self.market_context_snapshot_id, "market_context_snapshot_id") - _require_hash(self.market_context_logical_sha256, "market_context_logical_sha256") - schemas = tuple(self.event_schemas) - if not schemas or schemas != tuple(sorted(set(schemas))): - raise ValidationError("event_schemas must be non-empty, unique, and sorted") - if not isinstance(self.table, pa.Table) or self.table.num_rows <= 0: - raise ValidationError("verified input table must be a non-empty Arrow table") - if self.selection_logical_sha256 != _arrow_table_logical_sha256(self.table): - raise ValidationError("verified input selection hash does not match its Arrow table") - lineage = tuple(self.lineage) - if not lineage or lineage != tuple(sorted(lineage)): - raise ValidationError("lineage must be non-empty and canonically ordered") - lineage_keys = [(item.role, item.snapshot_id) for item in lineage] - if len(lineage_keys) != len(set(lineage_keys)): - raise ValidationError("lineage roles and snapshots must be unique") - source_lineage = [item for item in lineage if item.role == "market"] - if len(source_lineage) != 1 or ( - source_lineage[0].snapshot_id, - source_lineage[0].logical_sha256, - ) != (self.source_snapshot_id, self.source_logical_sha256): - raise ValidationError("source snapshot differs from its market lineage") - context_lineage = [item for item in lineage if item.role == "market_context"] - if len(context_lineage) != 1 or ( - context_lineage[0].snapshot_id, - context_lineage[0].logical_sha256, - ) != (self.market_context_snapshot_id, self.market_context_logical_sha256): - raise ValidationError("market context differs from its lineage") - if self.layer == "curated": - if self.aggregation is None: - raise ValidationError("Curated verified input requires aggregation metadata") - if schemas != (EventSchemaRef(BAR_EVENT_SCHEMA_ID, SCHEMA_VERSION_V2),): - raise ValidationError("Curated verified input requires the frozen Bar schema") - if self.table.schema != get_arrow_schema(BAR_EVENT_SCHEMA_ID): - raise ValidationError("Curated verified input table is not the frozen Bar schema") - context_values = ( - self.calendar_id, - self.session_policy_version, - self.market_context_snapshot_id, - self.market_context_logical_sha256, - ) - aggregation_values = ( - self.aggregation.calendar_id, - self.aggregation.session_policy_version, - self.aggregation.market_context_snapshot_id, - self.aggregation.market_context_logical_sha256, - ) - if context_values != aggregation_values: - raise ValidationError("verified input context differs from its aggregation") - elif self.aggregation is not None: - raise ValidationError("Normalized verified input cannot contain aggregation metadata") - elif any( - item.schema_id == BAR_EVENT_SCHEMA_ID or item.schema_version != SCHEMA_VERSION_V2 - for item in schemas - ): - raise ValidationError("Normalized verified input requires non-Bar v2 event schemas") - if self.layer == "normalized": - if "event_schema_id" not in self.table.column_names: - raise ValidationError("Normalized verified input lacks event_schema_id") - actual_schema_ids = set(self.table.column("event_schema_id").to_pylist()) - expected_schema_ids = {item.schema_id for item in schemas} - if actual_schema_ids != expected_schema_ids: - raise ValidationError("Normalized table event schemas differ from its contract") - object.__setattr__(self, "event_schemas", schemas) - object.__setattr__(self, "lineage", lineage) - - @classmethod - def _from_certified_factory(cls, **values: Any) -> VerifiedFactorInput: - return cls(_factory_token=_VERIFIED_INPUT_FACTORY_TOKEN, **values) + """Opaque immutable value created only after a complete certified loader succeeds.""" + + __slots__ = ( + "aggregation", + "calendar_id", + "event_schemas", + "layer", + "lineage", + "market_context_logical_sha256", + "market_context_snapshot_id", + "schema_id", + "selection_logical_sha256", + "session_policy_version", + "source_logical_sha256", + "source_snapshot_id", + "table", + ) + + def __new__(cls, *_args: Any, **_kwargs: Any) -> NoReturn: + raise ValidationError("VerifiedFactorInput can only be created by a certified loader") + + def __setattr__(self, _name: str, _value: Any) -> None: + raise ValidationError("VerifiedFactorInput is immutable") @property def arrow_schema_sha256(self) -> str: @@ -502,6 +428,97 @@ def to_contract(self) -> dict[str, Any]: } +def _validate_verified_factor_input_values(values: Mapping[str, Any]) -> None: + expected = set(VerifiedFactorInput.__slots__) + if set(values) != expected: + raise ValidationError("VerifiedFactorInput fields must be closed and complete") + schema_id = values["schema_id"] + layer = values["layer"] + source_snapshot_id = values["source_snapshot_id"] + source_logical_sha256 = values["source_logical_sha256"] + selection_logical_sha256 = values["selection_logical_sha256"] + event_schemas = values["event_schemas"] + table = values["table"] + calendar_id = values["calendar_id"] + session_policy_version = values["session_policy_version"] + market_context_snapshot_id = values["market_context_snapshot_id"] + market_context_logical_sha256 = values["market_context_logical_sha256"] + lineage = values["lineage"] + aggregation = values["aggregation"] + + if schema_id != VERIFIED_FACTOR_INPUT_SCHEMA_ID: + raise ValidationError("unsupported VerifiedFactorInput schema") + if layer not in {"curated", "normalized"}: + raise ValidationError("unsupported verified input layer") + _require_snapshot_id(source_snapshot_id, "source_snapshot_id") + _require_hash(source_logical_sha256, "source_logical_sha256") + _require_hash(selection_logical_sha256, "selection_logical_sha256") + _required_text(calendar_id, "calendar_id") + _required_text(session_policy_version, "session_policy_version") + _require_snapshot_id(market_context_snapshot_id, "market_context_snapshot_id") + _require_hash(market_context_logical_sha256, "market_context_logical_sha256") + schemas = tuple(event_schemas) + if not schemas or schemas != tuple(sorted(set(schemas))): + raise ValidationError("event_schemas must be non-empty, unique, and sorted") + if not isinstance(table, pa.Table) or table.num_rows <= 0: + raise ValidationError("verified input table must be a non-empty Arrow table") + if selection_logical_sha256 != _arrow_table_logical_sha256(table): + raise ValidationError("verified input selection hash does not match its Arrow table") + canonical_lineage = tuple(lineage) + if not canonical_lineage or canonical_lineage != tuple(sorted(canonical_lineage)): + raise ValidationError("lineage must be non-empty and canonically ordered") + lineage_keys = [(item.role, item.snapshot_id) for item in canonical_lineage] + if len(lineage_keys) != len(set(lineage_keys)): + raise ValidationError("lineage roles and snapshots must be unique") + source_lineage = [item for item in canonical_lineage if item.role == "market"] + if len(source_lineage) != 1 or ( + source_lineage[0].snapshot_id, + source_lineage[0].logical_sha256, + ) != (source_snapshot_id, source_logical_sha256): + raise ValidationError("source snapshot differs from its market lineage") + context_lineage = [item for item in canonical_lineage if item.role == "market_context"] + if len(context_lineage) != 1 or ( + context_lineage[0].snapshot_id, + context_lineage[0].logical_sha256, + ) != (market_context_snapshot_id, market_context_logical_sha256): + raise ValidationError("market context differs from its lineage") + if layer == "curated": + if aggregation is None: + raise ValidationError("Curated verified input requires aggregation metadata") + if schemas != (EventSchemaRef(BAR_EVENT_SCHEMA_ID, SCHEMA_VERSION_V2),): + raise ValidationError("Curated verified input requires the frozen Bar schema") + if table.schema != get_arrow_schema(BAR_EVENT_SCHEMA_ID): + raise ValidationError("Curated verified input table is not the frozen Bar schema") + context_values = ( + calendar_id, + session_policy_version, + market_context_snapshot_id, + market_context_logical_sha256, + ) + aggregation_values = ( + aggregation.calendar_id, + aggregation.session_policy_version, + aggregation.market_context_snapshot_id, + aggregation.market_context_logical_sha256, + ) + if context_values != aggregation_values: + raise ValidationError("verified input context differs from its aggregation") + elif aggregation is not None: + raise ValidationError("Normalized verified input cannot contain aggregation metadata") + elif any( + item.schema_id == BAR_EVENT_SCHEMA_ID or item.schema_version != SCHEMA_VERSION_V2 + for item in schemas + ): + raise ValidationError("Normalized verified input requires non-Bar v2 event schemas") + if layer == "normalized": + if "event_schema_id" not in table.column_names: + raise ValidationError("Normalized verified input lacks event_schema_id") + actual_schema_ids = set(table.column("event_schema_id").to_pylist()) + expected_schema_ids = {item.schema_id for item in schemas} + if actual_schema_ids != expected_schema_ids: + raise ValidationError("Normalized table event schemas differ from its contract") + + def _arrow_table_logical_sha256(table: pa.Table) -> str: combined = table.combine_chunks() sink = pa.BufferOutputStream() diff --git a/src/quant_data_kit/research_inputs_v2.py b/src/quant_data_kit/research_inputs_v2.py index 3c8ba73..447c700 100644 --- a/src/quant_data_kit/research_inputs_v2.py +++ b/src/quant_data_kit/research_inputs_v2.py @@ -48,11 +48,13 @@ from quant_data_kit.market_clock_v2 import MarketClock from quant_data_kit.research_contracts_v2 import ( MARKET_CONTEXT_SCHEMA_ID, + VERIFIED_FACTOR_INPUT_SCHEMA_ID, CuratedAggregation, EventBarPartitionEvidence, EventSchemaRef, LineageRef, VerifiedFactorInput, + _validate_verified_factor_input_values, ) from quant_data_kit.schemas_v2 import ( BAR_EVENT_SCHEMA_ID, @@ -830,18 +832,18 @@ def load_verified_normalized_events( if context_before.manifest() != context_after.manifest(): raise ValidationError("market context changed while building verified input") _assert_file_stamps(partition_stamps) - return VerifiedFactorInput._from_certified_factory( - layer="normalized", - source_snapshot_id=before.snapshot_id, - source_logical_sha256=before.logical_sha256, - selection_logical_sha256=_table_logical_sha256(table), - event_schemas=refs, - table=table, - calendar_id=context_before.calendar_id, - session_policy_version=context_before.session_policy_version, - market_context_snapshot_id=context_before.snapshot_id, - market_context_logical_sha256=context_before.logical_sha256, - lineage=tuple( + values = { + "layer": "normalized", + "source_snapshot_id": before.snapshot_id, + "source_logical_sha256": before.logical_sha256, + "selection_logical_sha256": _table_logical_sha256(table), + "event_schemas": refs, + "table": table, + "calendar_id": context_before.calendar_id, + "session_policy_version": context_before.session_policy_version, + "market_context_snapshot_id": context_before.snapshot_id, + "market_context_logical_sha256": context_before.logical_sha256, + "lineage": tuple( sorted( ( LineageRef("market", before.snapshot_id, before.logical_sha256), @@ -851,7 +853,14 @@ def load_verified_normalized_events( ) ) ), - ) + "aggregation": None, + "schema_id": VERIFIED_FACTOR_INPUT_SCHEMA_ID, + } + _validate_verified_factor_input_values(values) + verified = object.__new__(VerifiedFactorInput) + for name in VerifiedFactorInput.__slots__: + object.__setattr__(verified, name, values[name]) + return verified def _validate_bar_rows( @@ -1087,18 +1096,18 @@ def load_verified_curated_bars( if normalized_before != normalized: raise ValidationError("Normalized lineage changed while verifying event Bars") _assert_file_stamps(normalized_stamps) - return VerifiedFactorInput._from_certified_factory( - layer="curated", - source_snapshot_id=before.snapshot_id, - source_logical_sha256=before.logical_sha256, - selection_logical_sha256=_table_logical_sha256(table), - event_schemas=(EventSchemaRef(BAR_EVENT_SCHEMA_ID, SCHEMA_VERSION_V2),), - table=table, - calendar_id=aggregation.calendar_id, - session_policy_version=aggregation.session_policy_version, - market_context_snapshot_id=context_before.snapshot_id, - market_context_logical_sha256=context_before.logical_sha256, - lineage=tuple( + values = { + "layer": "curated", + "source_snapshot_id": before.snapshot_id, + "source_logical_sha256": before.logical_sha256, + "selection_logical_sha256": _table_logical_sha256(table), + "event_schemas": (EventSchemaRef(BAR_EVENT_SCHEMA_ID, SCHEMA_VERSION_V2),), + "table": table, + "calendar_id": aggregation.calendar_id, + "session_policy_version": aggregation.session_policy_version, + "market_context_snapshot_id": context_before.snapshot_id, + "market_context_logical_sha256": context_before.logical_sha256, + "lineage": tuple( sorted( ( LineageRef("market", before.snapshot_id, before.logical_sha256), @@ -1109,5 +1118,11 @@ def load_verified_curated_bars( ) ) ), - aggregation=aggregation, - ) + "aggregation": aggregation, + "schema_id": VERIFIED_FACTOR_INPUT_SCHEMA_ID, + } + _validate_verified_factor_input_values(values) + verified = object.__new__(VerifiedFactorInput) + for name in VerifiedFactorInput.__slots__: + object.__setattr__(verified, name, values[name]) + return verified diff --git a/tests/test_m2_integration.py b/tests/test_m2_integration.py index c1c001a..327dbdc 100644 --- a/tests/test_m2_integration.py +++ b/tests/test_m2_integration.py @@ -37,7 +37,7 @@ def crypto_context(provider: str) -> qdk.AdapterContext: def test_public_m2_api_and_version_are_exposed() -> None: - assert qdk.__version__ == version("quant-data-kit") == "0.8.0" + assert qdk.__version__ == version("quant-data-kit") == "0.8.1" for name in ( "write_raw_bytes", "write_normalized_events", diff --git a/tests/test_m8_research_contracts.py b/tests/test_m8_research_contracts.py index c840b62..79ccc6f 100644 --- a/tests/test_m8_research_contracts.py +++ b/tests/test_m8_research_contracts.py @@ -242,11 +242,17 @@ def valid_verified_input(**changes) -> VerifiedFactorInput: LineageRef("market", SNAPSHOT, HASH), LineageRef("market_context", SNAPSHOT, HASH), ), + "aggregation": None, + "schema_id": VERIFIED_FACTOR_INPUT_SCHEMA_ID, } values.update(changes) if "table" in changes and "selection_logical_sha256" not in changes: values["selection_logical_sha256"] = contracts._arrow_table_logical_sha256(values["table"]) - return VerifiedFactorInput._from_certified_factory(**values) + contracts._validate_verified_factor_input_values(values) + verified = object.__new__(VerifiedFactorInput) + for name in VerifiedFactorInput.__slots__: + object.__setattr__(verified, name, values[name]) + return verified def bar_table() -> pa.Table: @@ -303,7 +309,7 @@ def test_verified_input_is_closed_nonempty_and_layer_safe() -> None: table=curated_table, ) assert curated.to_contract()["aggregation"]["kind"] == "fixed_time_bar" - with pytest.raises(ValidationError, match="certified factory"): + with pytest.raises(ValidationError, match="certified loader"): VerifiedFactorInput( layer="normalized", source_snapshot_id=SNAPSHOT, @@ -317,7 +323,9 @@ def test_verified_input_is_closed_nonempty_and_layer_safe() -> None: market_context_logical_sha256=HASH, lineage=(LineageRef("market", SNAPSHOT, HASH),), ) - with pytest.raises(ValidationError, match="certified factory"): + assert not hasattr(contracts, "_VERIFIED_INPUT_FACTORY_TOKEN") + assert not hasattr(VerifiedFactorInput, "_from_certified_factory") + with pytest.raises(TypeError, match="dataclass"): replace(valid, source_snapshot_id=f"sha256-{OTHER_HASH}") with pytest.raises(ValidationError, match="market lineage"): valid_verified_input(source_snapshot_id=f"sha256-{OTHER_HASH}")