fix: type-check factor set definition ids (#16)

MODULE_SPEC.yaml

@@ -1,7 +1,7 @@
 {
   "schema_version": 1,
   "module_id": "quant_engine",
-  "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 1, "effective_from": "2026-08-20T00:00:00+08:00"},
+  "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 2, "effective_from": "2026-09-01T00:00:00+08:00"},
   "repository": {"name": "quant_engine", "workspace_id": "researchhub", "type": "research_engine", "maturity": "operational"},
   "bounded_context": {
     "domain": "quantitative-research-engine",
@@ -24,7 +24,16 @@
     {"asset_id": "quantitative-model-implementations", "kind": "model", "classification": "internal"},
     {"asset_id": "simulation-and-metric-results", "kind": "artifact", "classification": "confidential"}
   ]},
-  "contracts": {"provides": [], "consumes": []},
+  "contracts": {
+    "provides": [
+      {"contract_id": "researchhub.factor-definition", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/factor_contracts.py"},
+      {"contract_id": "researchhub.factor-set-ref", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/factor_contracts.py"}
+    ],
+    "consumes": [
+      {"contract_id": "researchhub.dataset-snapshot", "version": "1.0.0", "authority": "researchhub.data", "admission": "qualified_immutable_envelope"},
+      {"contract_id": "researchhub.data-foundation", "version": "1.0.0", "authority": "researchhub.data", "admission": "content_addressed_selected_views"}
+    ]
+  },
   "dependencies": [],
   "agent_context": {
     "default_entrypoints": [

README.md

@@ -19,12 +19,14 @@
 ## 模块
 
 - `alpha_factors` — 158 alpha 公式 + 24 基础算子（移植自 qlib alpha158）
+- `factor_contracts` — `FactorDefinition` / `FactorSetRef` v1 纯计算合同、严格 PIT/availability 输入准入与显式 legacy 投影
 - `execution` — A 股长仓执行仿真（成本/滑点/现金约束）+ 稀疏调仓/完整交易日 Ledger + 可投影成交与 NAV 审计；T+1、涨跌停、成交量与价差提供独立约束函数
 - `indicators` — 50+ 技术指标（MACD / KDJ / 布林 / ATR / ADX / 等）
 - `data_adapter` — 桥接 qtdb_pro 长表与新模块（rename / long-wide / 复权 / vwap 代理）
 - `backtest` — weight-based 多日仿真（rebalance_table / compute_nav / compare_to_benchmark）
 - `portfolio_construction` — 多期因子分数 → Top-K → 等权目标权重表
 - `research_pipeline` — 因子日 → 下一真实交易日 → 显式执行价 → 日末估值 → 成本后绩效（防前视编排）
+- `governed_pipeline` — 数据快照 → 因子版本 → 策略版本 → 回测运行 → 目标组合 → 风险决策 → Paper 订单意图；全链路带确定性 ID，风险拒绝时禁止生成订单意图
 - `artifact` — 版本化、确定性、存储中立的完整 research run 事实表与 manifest
 - `attribution` — 基于实际成交后持仓的隔夜 / 日内 / 交易成本逐日收益归因与闭合审计
 - `metrics` — 绝对绩效 + 严格日期对齐的 TE / IR / alpha / beta 基准相对绩效
@@ -55,6 +57,9 @@ pytest                    # 单元测试
 pytest --cov=src          # 覆盖率
 mypy --strict src/        # 类型检查
 ruff check src/ tests/    # lint
+
+# 无网络、无数据库、无券商的架构烟测
+uv run python -m quant_engine.governed_pipeline
 ```
 
 ## 使用
@@ -191,6 +196,68 @@ print(backtest.stats())
 print(backtest.benchmark_report())
 ```
 
+## 因子／特征合同 v1
+
+`quant_engine.factor_contracts` 提供 `researchhub.factor-definition` 与
+`researchhub.factor-set-ref` `1.0.0`。合同使用受限 canonical JSON：只接受 ASCII
+lower-snake-case object key、UTF-8 string、bool/null 和 safe integer；小数参数必须用显式
+canonical decimal string。定义、输入映射、上游证据、输出 schema/content 和 lineage 的任一
+语义变化都会产生新 identity。
+
+创建 `FactorSetRef` 必须提供完整且可重算 identity 的 `DatasetSnapshotEnvelope` 与
+`DataFoundationEnvelope`，不能用 ID 字符串或布尔值代替资格证明。每个因子输入都要映射到一个
+实际选中的 `StandardizedViewRef`，schema 必须同时匹配定义和 view；未消费、缺失、重复或跨
+snapshot/Foundation/PIT 的 view 都会失败关闭。snapshot PIT 可以早于 Foundation/view PIT，
+但始终满足 knowledge ≤ snapshot PIT ≤ Foundation/view/FactorSet PIT ≤ evaluation。
+
+```python
+from quant_engine.factor_contracts import (
+    DataFoundationEnvelope,
+    DatasetSnapshotEnvelope,
+    FactorDefinition,
+    FactorSetRef,
+)
+
+snapshot = DatasetSnapshotEnvelope.from_dict(dataset_snapshot_v1)
+foundation = DataFoundationEnvelope.from_dict(data_foundation_v1)
+
+# definition 必须是完整的 FactorDefinition；FactorSetRef.create 还要求显式 input bindings、
+# view availability、output quality/coverage、canonical output bytes 和 immutable artifact ref。
+factor_set = FactorSetRef.create(
+    definitions=(definition,),
+    dataset_snapshot=snapshot,
+    foundation=foundation,
+    **explicit_factor_set_evidence,
+)
+```
+
+`availability_mode="as_available"` 声明 source/view 和计算产物在历史 evaluation 前实际可用；
+`"retrospective_replay"` 保留历史 evaluation，但要求真实 publication/view creation、compute 和
+artifact 时间位于之后，并固定 `historical_availability="not_established"`。两种模式都不会授予
+decision、real-data、production、paper 或 live readiness。
+
+旧 `governed_pipeline.FactorVersion` 的四字段构造器、`factor_id@version`、run/target/risk/order
+identity 均保持不变。迁移只能通过 content-addressed `LegacyFactorBinding`，再显式调用
+`bind_legacy_factor()` 或 `project_legacy_factor()`；后者是有损投影，不表示旧 digest 与新定义
+digest 等价，也不会把旧 run 静默升级为新合同。
+
+## 治理垂直切片
+
+`governed_pipeline` 不复制因子、回测、组合或执行算法，只编排现有能力并补充版本与风险契约。
+调用方必须显式提供 `DatasetSnapshot`、`FactorVersion`、`StrategyVersion`、代码提交和
+`RiskPolicy`。模块只会生成 `environment="paper"` 的订单意图，不连接数据库、数据供应商或
+券商；风险决策为拒绝时，订单意图固定为空，直接调用创建函数也会失败关闭。
+
+该切片对应 ResearchHub 架构的首个可执行验收链路：
+
+```text
+DatasetSnapshot → FactorVersion → StrategyVersion → BacktestRun
+→ PortfolioTarget → RiskDecision → PaperOrderIntent
+```
+
+平台总架构、五仓职责和十二层能力映射仍以 `research_platform/docs/architecture/` 为权威；
+本仓只拥有纯计算与离线模拟合同。
+
 ## 与 research_results 的关系
 
 `research_results` 依赖 `quant_engine`（通过 re-export 保持向后兼容）：

src/quant_engine/alpha_factors.py

@@ -3271,6 +3271,363 @@ def evaluate_phase3_formula(name: str, **inputs: pd.Series) -> pd.Series:
     return function(*(inputs[field] for field in required_inputs))
 
 
+# ── Phase 4 formula contract: frozen alpha051-alpha100 surface ──────────────
+
+# Phase 4 extends the versioned formula contract without mutating the Phase 3
+# catalogue, digest, dispatch surface, or the existing formula functions.
+ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION = "1.0.0"
+ALPHA158_PHASE4_FORMULA_CATALOG_SHA256 = (
+    "858daf5e2abab5063fc28fcf7c79936096e2c76dde582fc7bab78b3458f17054"
+)
+
+_PHASE4_FORMULA_FUNCTIONS: dict[str, Callable[..., pd.Series]] = {
+    "alpha_051": alpha_051,
+    "alpha_052": alpha_052,
+    "alpha_053": alpha_053,
+    "alpha_054": alpha_054,
+    "alpha_055": alpha_055,
+    "alpha_056": alpha_056,
+    "alpha_057": alpha_057,
+    "alpha_058": alpha_058,
+    "alpha_059": alpha_059,
+    "alpha_060": alpha_060,
+    "alpha_061": alpha_061,
+    "alpha_062": alpha_062,
+    "alpha_063": alpha_063,
+    "alpha_064": alpha_064,
+    "alpha_065": alpha_065,
+    "alpha_066": alpha_066,
+    "alpha_067": alpha_067,
+    "alpha_068": alpha_068,
+    "alpha_069": alpha_069,
+    "alpha_070": alpha_070,
+    "alpha_071": alpha_071,
+    "alpha_072": alpha_072,
+    "alpha_073": alpha_073,
+    "alpha_074": alpha_074,
+    "alpha_075": alpha_075,
+    "alpha_076": alpha_076,
+    "alpha_077": alpha_077,
+    "alpha_078": alpha_078,
+    "alpha_079": alpha_079,
+    "alpha_080": alpha_080,
+    "alpha_081": alpha_081,
+    "alpha_082": alpha_082,
+    "alpha_083": alpha_083,
+    "alpha_084": alpha_084,
+    "alpha_085": alpha_085,
+    "alpha_086": alpha_086,
+    "alpha_087": alpha_087,
+    "alpha_088": alpha_088,
+    "alpha_089": alpha_089,
+    "alpha_090": alpha_090,
+    "alpha_091": alpha_091,
+    "alpha_092": alpha_092,
+    "alpha_093": alpha_093,
+    "alpha_094": alpha_094,
+    "alpha_095": alpha_095,
+    "alpha_096": alpha_096,
+    "alpha_097": alpha_097,
+    "alpha_098": alpha_098,
+    "alpha_099": alpha_099,
+    "alpha_100": alpha_100,
+}
+
+_PHASE4_INPUT_CATEGORIES = {
+    1: "single",
+    2: "pair",
+    3: "triple",
+    4: "quadruple",
+    5: "quintuple",
+}
+
+
+def _build_phase4_formula_specs() -> dict[str, dict[str, Any]]:
+    specs: dict[str, dict[str, Any]] = {}
+    for alpha_id, function in _PHASE4_FORMULA_FUNCTIONS.items():
+        meta = ALPHA158_REGISTRY[alpha_id]
+        call_inputs = _phase3_call_inputs(function)
+        formula_inputs = _phase3_string_list(meta, "inputs", alpha_id)
+        input_category = _PHASE4_INPUT_CATEGORIES.get(len(call_inputs))
+        if input_category is None:
+            raise RuntimeError(f"unsupported formula input count for {alpha_id}")
+        specs[alpha_id] = {
+            "name": alpha_id,
+            "contract_version": ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION,
+            "formula": meta["formula"],
+            "category": meta["category"],
+            "complexity": meta["complexity"],
+            "parameters": _phase3_string_list(meta, "params", alpha_id),
+            "description": meta["description"],
+            "references": _phase3_string_list(meta, "references", alpha_id),
+            "call_inputs": list(call_inputs),
+            "formula_inputs": formula_inputs,
+            "input_category": input_category,
+        }
+    return specs
+
+
+ALPHA158_PHASE4_FORMULA_SPECS: Mapping[str, Mapping[str, Any]] = (
+    _freeze_phase3_formula_specs(_build_phase4_formula_specs())
+)
+
+
+def list_phase4_formulas() -> tuple[str, ...]:
+    """Return the frozen alpha051-alpha100 formula IDs in stable order."""
+    return tuple(ALPHA158_PHASE4_FORMULA_SPECS)
+
+
+def evaluate_phase4_formula(name: str, **inputs: pd.Series) -> pd.Series:
+    """Evaluate a Phase 4 formula with an exact, alignment-safe input contract."""
+    if name not in ALPHA158_PHASE4_FORMULA_SPECS:
+        raise KeyError(f"formula {name!r} not registered")
+
+    spec = ALPHA158_PHASE4_FORMULA_SPECS[name]
+    required_inputs = cast(tuple[str, ...], spec["call_inputs"])
+    missing_inputs = [field for field in required_inputs if field not in inputs]
+    unexpected_inputs = sorted(field for field in inputs if field not in required_inputs)
+    if missing_inputs or unexpected_inputs:
+        details: list[str] = []
+        if missing_inputs:
+            details.append(f"missing inputs {missing_inputs}")
+        if unexpected_inputs:
+            details.append(f"unexpected inputs {unexpected_inputs}")
+        raise ValueError(f"invalid inputs for {name}: {'; '.join(details)}")
+
+    for field in required_inputs:
+        if not isinstance(inputs[field], pd.Series):
+            raise TypeError(f"{field} must be a pandas Series")
+
+    primary_field = required_inputs[0]
+    primary = inputs[primary_field]
+    for field in required_inputs[1:]:
+        if not primary.index.equals(inputs[field].index):
+            raise ValueError(f"{field} index must align with {primary_field}")
+
+    function = _PHASE4_FORMULA_FUNCTIONS[name]
+    return function(*(inputs[field] for field in required_inputs))
+
+
+# ── Phase 5 formula contract: frozen alpha101-alpha150 surface ──────────────
+
+# Phase 5 extends the versioned formula contract without mutating any earlier
+# catalogue, digest, dispatch surface, or existing formula implementation.
+ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION = "1.0.0"
+ALPHA158_PHASE5_FORMULA_CATALOG_SHA256 = (
+    "3368796169c9fbd39c4a34ea137e569964b15882fbf4de25124790d548db6533"
+)
+
+_PHASE5_FORMULA_FUNCTIONS: dict[str, Callable[..., pd.Series]] = {
+    "alpha_101": alpha_101,
+    "alpha_102": alpha_102,
+    "alpha_103": alpha_103,
+    "alpha_104": alpha_104,
+    "alpha_105": alpha_105,
+    "alpha_106": alpha_106,
+    "alpha_107": alpha_107,
+    "alpha_108": alpha_108,
+    "alpha_109": alpha_109,
+    "alpha_110": alpha_110,
+    "alpha_111": alpha_111,
+    "alpha_112": alpha_112,
+    "alpha_113": alpha_113,
+    "alpha_114": alpha_114,
+    "alpha_115": alpha_115,
+    "alpha_116": alpha_116,
+    "alpha_117": alpha_117,
+    "alpha_118": alpha_118,
+    "alpha_119": alpha_119,
+    "alpha_120": alpha_120,
+    "alpha_121": alpha_121,
+    "alpha_122": alpha_122,
+    "alpha_123": alpha_123,
+    "alpha_124": alpha_124,
+    "alpha_125": alpha_125,
+    "alpha_126": alpha_126,
+    "alpha_127": alpha_127,
+    "alpha_128": alpha_128,
+    "alpha_129": alpha_129,
+    "alpha_130": alpha_130,
+    "alpha_131": alpha_131,
+    "alpha_132": alpha_132,
+    "alpha_133": alpha_133,
+    "alpha_134": alpha_134,
+    "alpha_135": alpha_135,
+    "alpha_136": alpha_136,
+    "alpha_137": alpha_137,
+    "alpha_138": alpha_138,
+    "alpha_139": alpha_139,
+    "alpha_140": alpha_140,
+    "alpha_141": alpha_141,
+    "alpha_142": alpha_142,
+    "alpha_143": alpha_143,
+    "alpha_144": alpha_144,
+    "alpha_145": alpha_145,
+    "alpha_146": alpha_146,
+    "alpha_147": alpha_147,
+    "alpha_148": alpha_148,
+    "alpha_149": alpha_149,
+    "alpha_150": alpha_150,
+}
+
+
+def _build_phase5_formula_specs() -> dict[str, dict[str, Any]]:
+    specs: dict[str, dict[str, Any]] = {}
+    for alpha_id, function in _PHASE5_FORMULA_FUNCTIONS.items():
+        meta = ALPHA158_REGISTRY[alpha_id]
+        call_inputs = _phase3_call_inputs(function)
+        formula_inputs = _phase3_string_list(meta, "inputs", alpha_id)
+        input_category = _PHASE4_INPUT_CATEGORIES.get(len(call_inputs))
+        if input_category is None:
+            raise RuntimeError(f"unsupported formula input count for {alpha_id}")
+        specs[alpha_id] = {
+            "name": alpha_id,
+            "contract_version": ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION,
+            "formula": meta["formula"],
+            "category": meta["category"],
+            "complexity": meta["complexity"],
+            "parameters": _phase3_string_list(meta, "params", alpha_id),
+            "description": meta["description"],
+            "references": _phase3_string_list(meta, "references", alpha_id),
+            "call_inputs": list(call_inputs),
+            "formula_inputs": formula_inputs,
+            "input_category": input_category,
+        }
+    return specs
+
+
+ALPHA158_PHASE5_FORMULA_SPECS: Mapping[str, Mapping[str, Any]] = (
+    _freeze_phase3_formula_specs(_build_phase5_formula_specs())
+)
+
+
+def list_phase5_formulas() -> tuple[str, ...]:
+    """Return the frozen alpha101-alpha150 formula IDs in stable order."""
+    return tuple(ALPHA158_PHASE5_FORMULA_SPECS)
+
+
+def evaluate_phase5_formula(name: str, **inputs: pd.Series) -> pd.Series:
+    """Evaluate a Phase 5 formula with an exact, alignment-safe input contract."""
+    if name not in ALPHA158_PHASE5_FORMULA_SPECS:
+        raise KeyError(f"formula {name!r} not registered")
+
+    spec = ALPHA158_PHASE5_FORMULA_SPECS[name]
+    required_inputs = cast(tuple[str, ...], spec["call_inputs"])
+    missing_inputs = [field for field in required_inputs if field not in inputs]
+    unexpected_inputs = sorted(field for field in inputs if field not in required_inputs)
+    if missing_inputs or unexpected_inputs:
+        details: list[str] = []
+        if missing_inputs:
+            details.append(f"missing inputs {missing_inputs}")
+        if unexpected_inputs:
+            details.append(f"unexpected inputs {unexpected_inputs}")
+        raise ValueError(f"invalid inputs for {name}: {'; '.join(details)}")
+
+    for field in required_inputs:
+        if not isinstance(inputs[field], pd.Series):
+            raise TypeError(f"{field} must be a pandas Series")
+
+    primary_field = required_inputs[0]
+    primary = inputs[primary_field]
+    for field in required_inputs[1:]:
+        if len(inputs[field]) != len(primary):
+            raise ValueError(f"{field} length must match {primary_field}")
+        if not primary.index.equals(inputs[field].index):
+            raise ValueError(f"{field} index must align with {primary_field}")
+
+    function = _PHASE5_FORMULA_FUNCTIONS[name]
+    return function(*(inputs[field] for field in required_inputs))
+
+
+# ── Phase 6 formula contract: frozen alpha151-alpha158 surface ──────────────
+
+# Phase 6 completes the versioned formula contract without mutating any
+# earlier catalogue, digest, dispatch surface, or formula implementation.
+ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION = "1.0.0"
+ALPHA158_PHASE6_FORMULA_CATALOG_SHA256 = (
+    "70ffae16ca6cbb59a1e8644f5cdeec1d291fa75ff8b5647fb534af0083408219"
+)
+
+_PHASE6_FORMULA_FUNCTIONS: dict[str, Callable[..., pd.Series]] = {
+    "alpha_151": alpha_151,
+    "alpha_152": alpha_152,
+    "alpha_153": alpha_153,
+    "alpha_154": alpha_154,
+    "alpha_155": alpha_155,
+    "alpha_156": alpha_156,
+    "alpha_157": alpha_157,
+    "alpha_158": alpha_158,
+}
+
+
+def _build_phase6_formula_specs() -> dict[str, dict[str, Any]]:
+    specs: dict[str, dict[str, Any]] = {}
+    for alpha_id, function in _PHASE6_FORMULA_FUNCTIONS.items():
+        meta = ALPHA158_REGISTRY[alpha_id]
+        call_inputs = _phase3_call_inputs(function)
+        formula_inputs = _phase3_string_list(meta, "inputs", alpha_id)
+        input_category = _PHASE4_INPUT_CATEGORIES.get(len(call_inputs))
+        if input_category is None:
+            raise RuntimeError(f"unsupported formula input count for {alpha_id}")
+        specs[alpha_id] = {
+            "name": alpha_id,
+            "contract_version": ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION,
+            "formula": meta["formula"],
+            "category": meta["category"],
+            "complexity": meta["complexity"],
+            "parameters": _phase3_string_list(meta, "params", alpha_id),
+            "description": meta["description"],
+            "references": _phase3_string_list(meta, "references", alpha_id),
+            "call_inputs": list(call_inputs),
+            "formula_inputs": formula_inputs,
+            "input_category": input_category,
+        }
+    return specs
+
+
+ALPHA158_PHASE6_FORMULA_SPECS: Mapping[str, Mapping[str, Any]] = (
+    _freeze_phase3_formula_specs(_build_phase6_formula_specs())
+)
+
+
+def list_phase6_formulas() -> tuple[str, ...]:
+    """Return the frozen alpha151-alpha158 formula IDs in stable order."""
+    return tuple(ALPHA158_PHASE6_FORMULA_SPECS)
+
+
+def evaluate_phase6_formula(name: str, **inputs: pd.Series) -> pd.Series:
+    """Evaluate a Phase 6 formula with an exact, alignment-safe input contract."""
+    if name not in ALPHA158_PHASE6_FORMULA_SPECS:
+        raise KeyError(f"formula {name!r} not registered")
+
+    spec = ALPHA158_PHASE6_FORMULA_SPECS[name]
+    required_inputs = cast(tuple[str, ...], spec["call_inputs"])
+    missing_inputs = [field for field in required_inputs if field not in inputs]
+    unexpected_inputs = sorted(field for field in inputs if field not in required_inputs)
+    if missing_inputs or unexpected_inputs:
+        details: list[str] = []
+        if missing_inputs:
+            details.append(f"missing inputs {missing_inputs}")
+        if unexpected_inputs:
+            details.append(f"unexpected inputs {unexpected_inputs}")
+        raise ValueError(f"invalid inputs for {name}: {'; '.join(details)}")
+
+    for field in required_inputs:
+        if not isinstance(inputs[field], pd.Series):
+            raise TypeError(f"{field} must be a pandas Series")
+
+    primary_field = required_inputs[0]
+    primary = inputs[primary_field]
+    for field in required_inputs[1:]:
+        if len(inputs[field]) != len(primary):
+            raise ValueError(f"{field} length must match {primary_field}")
+        if not primary.index.equals(inputs[field].index):
+            raise ValueError(f"{field} index must align with {primary_field}")
+
+    function = _PHASE6_FORMULA_FUNCTIONS[name]
+    return function(*(inputs[field] for field in required_inputs))
+
+
 __all__ = [
     "rank",
     "delta",
@@ -3309,6 +3666,21 @@ __all__ = [
     "ALPHA158_PHASE3_FORMULA_SPECS",
     "list_phase3_formulas",
     "evaluate_phase3_formula",
+    "ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION",
+    "ALPHA158_PHASE4_FORMULA_CATALOG_SHA256",
+    "ALPHA158_PHASE4_FORMULA_SPECS",
+    "list_phase4_formulas",
+    "evaluate_phase4_formula",
+    "ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION",
+    "ALPHA158_PHASE5_FORMULA_CATALOG_SHA256",
+    "ALPHA158_PHASE5_FORMULA_SPECS",
+    "list_phase5_formulas",
+    "evaluate_phase5_formula",
+    "ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION",
+    "ALPHA158_PHASE6_FORMULA_CATALOG_SHA256",
+    "ALPHA158_PHASE6_FORMULA_SPECS",
+    "list_phase6_formulas",
+    "evaluate_phase6_formula",
     "alpha_001",
     "alpha_002",
     "alpha_003",

src/quant_engine/governed_pipeline.py

@@ -0,0 +1,622 @@
+"""Versioned, risk-gated, paper-only quantitative research vertical slice.
+
+The module composes existing ``quant_engine`` calculations into a small set of
+storage-neutral governance contracts.  It never reads a database, calls a data
+vendor, loads credentials, or routes an order to a broker.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import re
+from collections.abc import Mapping
+from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
+from enum import StrEnum
+from types import MappingProxyType
+
+import pandas as pd
+
+from quant_engine.execution import ExecutionConfig
+from quant_engine.factor_contracts import (
+    ContractErrorCode,
+    FactorContractError,
+    FactorDefinition,
+    LegacyFactorBinding,
+)
+from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research
+
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+_GIT_SHA = re.compile(r"^[0-9a-f]{40}$")
+
+__all__ = [
+    "DatasetSnapshot",
+    "FactorVersion",
+    "bind_legacy_factor",
+    "project_legacy_factor",
+    "StrategyStage",
+    "StrategyVersion",
+    "BacktestRun",
+    "PortfolioTarget",
+    "RiskPolicy",
+    "RiskDecisionStatus",
+    "RiskDecision",
+    "PaperOrderIntent",
+    "GovernedFactorSliceResult",
+    "evaluate_portfolio_risk",
+    "create_paper_order_intent",
+    "run_governed_factor_slice",
+]
+
+
+def _required_text(value: str, name: str) -> str:
+    normalized = value.strip()
+    if not normalized:
+        raise ValueError(f"{name} must be non-empty")
+    return normalized
+
+
+def _aware_utc(value: datetime, name: str) -> datetime:
+    if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError(f"{name} must be timezone-aware")
+    return value.astimezone(UTC)
+
+
+def _canonical_value(value: object) -> object:
+    if value is None or isinstance(value, str | bool | int):
+        return value
+    if isinstance(value, float):
+        if math.isnan(value):
+            return "NaN"
+        if math.isinf(value):
+            return "Infinity" if value > 0 else "-Infinity"
+        return value
+    if isinstance(value, datetime):
+        return _aware_utc(value, "datetime").isoformat()
+    if isinstance(value, StrEnum):
+        return value.value
+    if isinstance(value, Mapping):
+        return {
+            str(key): _canonical_value(item)
+            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
+        }
+    if isinstance(value, list | tuple):
+        return [_canonical_value(item) for item in value]
+    raise TypeError(f"unsupported canonical value: {type(value).__name__}")
+
+
+def _stable_id(prefix: str, payload: Mapping[str, object]) -> str:
+    encoded = json.dumps(
+        _canonical_value(payload),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode("utf-8")
+    return f"{prefix}:{hashlib.sha256(encoded).hexdigest()}"
+
+
+def _immutable_weights(values: Mapping[str, float]) -> Mapping[str, float]:
+    normalized: dict[str, float] = {}
+    for raw_asset, raw_weight in values.items():
+        asset = _required_text(str(raw_asset), "asset_id")
+        weight = float(raw_weight)
+        if not math.isfinite(weight) or weight < 0:
+            raise ValueError("portfolio weights must be finite and non-negative")
+        if asset in normalized:
+            raise ValueError(f"duplicate asset_id: {asset}")
+        normalized[asset] = weight
+    return MappingProxyType(dict(sorted(normalized.items())))
+
+
+@dataclass(frozen=True, slots=True)
+class DatasetSnapshot:
+    """Point-in-time identity for caller-supplied research data."""
+
+    snapshot_id: str
+    schema_version: str
+    content_sha256: str
+    effective_at: datetime
+    available_at: datetime
+    ingested_at: datetime
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "snapshot_id", _required_text(self.snapshot_id, "snapshot_id"))
+        object.__setattr__(
+            self, "schema_version", _required_text(self.schema_version, "schema_version")
+        )
+        digest = self.content_sha256.strip().lower()
+        if not _SHA256.fullmatch(digest):
+            raise ValueError("content_sha256 must be a lowercase SHA-256 digest")
+        object.__setattr__(self, "content_sha256", digest)
+        effective_at = _aware_utc(self.effective_at, "effective_at")
+        available_at = _aware_utc(self.available_at, "available_at")
+        ingested_at = _aware_utc(self.ingested_at, "ingested_at")
+        if not effective_at <= available_at <= ingested_at:
+            raise ValueError("timestamps must satisfy effective_at <= available_at <= ingested_at")
+        object.__setattr__(self, "effective_at", effective_at)
+        object.__setattr__(self, "available_at", available_at)
+        object.__setattr__(self, "ingested_at", ingested_at)
+
+
+@dataclass(frozen=True, slots=True)
+class FactorVersion:
+    """Versioned factor definition identity without factor implementation duplication."""
+
+    factor_id: str
+    version: str
+    definition_sha256: str
+    dataset_schema_version: str
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "factor_id", _required_text(self.factor_id, "factor_id"))
+        object.__setattr__(self, "version", _required_text(self.version, "version"))
+        object.__setattr__(
+            self,
+            "dataset_schema_version",
+            _required_text(self.dataset_schema_version, "dataset_schema_version"),
+        )
+        digest = self.definition_sha256.strip().lower()
+        if not _SHA256.fullmatch(digest):
+            raise ValueError("definition_sha256 must be a lowercase SHA-256 digest")
+        object.__setattr__(self, "definition_sha256", digest)
+
+    @property
+    def version_id(self) -> str:
+        return f"{self.factor_id}@{self.version}"
+
+
+def _validate_legacy_factor_binding(
+    legacy: FactorVersion,
+    definition: FactorDefinition,
+    binding: LegacyFactorBinding,
+) -> None:
+    if not isinstance(legacy, FactorVersion):
+        raise FactorContractError(
+            ContractErrorCode.TYPE_ERROR,
+            "$.legacy",
+            "FactorVersion is required",
+        )
+    if not isinstance(definition, FactorDefinition):
+        raise FactorContractError(
+            ContractErrorCode.TYPE_ERROR,
+            "$.definition",
+            "complete FactorDefinition is required",
+        )
+    if not isinstance(binding, LegacyFactorBinding):
+        raise FactorContractError(
+            ContractErrorCode.TYPE_ERROR,
+            "$.binding",
+            "LegacyFactorBinding is required",
+        )
+    expected = (
+        definition.definition_id,
+        legacy.factor_id,
+        legacy.version,
+        legacy.definition_sha256,
+        legacy.dataset_schema_version,
+        definition.input_schema_digest,
+    )
+    actual = (
+        binding.definition_id,
+        binding.legacy_factor_id,
+        binding.legacy_version,
+        binding.legacy_definition_sha256,
+        binding.legacy_dataset_schema_version,
+        binding.canonical_input_schema_digest,
+    )
+    if actual != expected:
+        raise FactorContractError(
+            ContractErrorCode.LEGACY_BINDING_MISMATCH,
+            "$.binding",
+            "binding does not exactly associate the supplied legacy and canonical identities",
+        )
+
+
+def bind_legacy_factor(
+    legacy: FactorVersion,
+    complete_definition: FactorDefinition,
+    binding: LegacyFactorBinding,
+) -> FactorDefinition:
+    """Validate an explicit migration binding without manufacturing missing semantics."""
+
+    _validate_legacy_factor_binding(legacy, complete_definition, binding)
+    return complete_definition
+
+
+def project_legacy_factor(
+    complete_definition: FactorDefinition,
+    binding: LegacyFactorBinding,
+) -> FactorVersion:
+    """Project a canonical definition into its recorded, explicitly lossy legacy identity."""
+
+    legacy = FactorVersion(
+        factor_id=binding.legacy_factor_id,
+        version=binding.legacy_version,
+        definition_sha256=binding.legacy_definition_sha256,
+        dataset_schema_version=binding.legacy_dataset_schema_version,
+    )
+    _validate_legacy_factor_binding(legacy, complete_definition, binding)
+    return legacy
+
+
+class StrategyStage(StrEnum):
+    DRAFT = "Draft"
+    RESEARCH = "Research"
+    VALIDATED = "Validated"
+    APPROVED = "Approved"
+    PAPER = "Paper"
+    LIVE = "Live"
+    PAUSED = "Paused"
+    RETIRED = "Retired"
+
+
+@dataclass(frozen=True, slots=True)
+class StrategyVersion:
+    """Strategy lineage and lifecycle state used by the governed slice."""
+
+    strategy_id: str
+    version: str
+    factor_version_id: str
+    stage: StrategyStage
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "strategy_id", _required_text(self.strategy_id, "strategy_id"))
+        object.__setattr__(self, "version", _required_text(self.version, "version"))
+        object.__setattr__(
+            self,
+            "factor_version_id",
+            _required_text(self.factor_version_id, "factor_version_id"),
+        )
+        if not isinstance(self.stage, StrategyStage):
+            raise TypeError("stage must be a StrategyStage")
+
+    @property
+    def version_id(self) -> str:
+        return f"{self.strategy_id}@{self.version}"
+
+
+@dataclass(frozen=True, slots=True)
+class BacktestRun:
+    run_id: str
+    dataset_snapshot_id: str
+    factor_version_id: str
+    strategy_version_id: str
+    code_revision: str
+    config_hash: str
+    created_at: datetime
+
+
+@dataclass(frozen=True, slots=True)
+class PortfolioTarget:
+    target_id: str
+    backtest_run_id: str
+    dataset_snapshot_id: str
+    weights: Mapping[str, float]
+    created_at: datetime
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "weights", _immutable_weights(self.weights))
+
+    @property
+    def gross_exposure(self) -> float:
+        return float(sum(abs(weight) for weight in self.weights.values()))
+
+
+@dataclass(frozen=True, slots=True)
+class RiskPolicy:
+    policy_id: str
+    max_gross_exposure: float
+    max_single_asset_weight: float
+    max_positions: int
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "policy_id", _required_text(self.policy_id, "policy_id"))
+        if not math.isfinite(self.max_gross_exposure) or self.max_gross_exposure <= 0:
+            raise ValueError("max_gross_exposure must be positive and finite")
+        if not math.isfinite(self.max_single_asset_weight) or not (
+            0 < self.max_single_asset_weight <= 1
+        ):
+            raise ValueError("max_single_asset_weight must be in (0, 1]")
+        if (
+            isinstance(self.max_positions, bool)
+            or not isinstance(self.max_positions, int)
+            or self.max_positions <= 0
+        ):
+            raise ValueError("max_positions must be a positive integer")
+
+
+class RiskDecisionStatus(StrEnum):
+    APPROVED = "approved"
+    REJECTED = "rejected"
+
+
+@dataclass(frozen=True, slots=True)
+class RiskDecision:
+    decision_id: str
+    portfolio_target_id: str
+    policy_id: str
+    status: RiskDecisionStatus
+    reasons: tuple[str, ...]
+    created_at: datetime
+
+
+@dataclass(frozen=True, slots=True, init=False)
+class PaperOrderIntent:
+    intent_id: str
+    portfolio_target_id: str
+    risk_decision_id: str
+    environment: str
+    target_weights: Mapping[str, float]
+    created_at: datetime
+
+    def __init__(self, target: PortfolioTarget, decision: RiskDecision) -> None:
+        if decision.status is not RiskDecisionStatus.APPROVED:
+            raise ValueError("an approved risk decision is required")
+        if decision.portfolio_target_id != target.target_id:
+            raise ValueError("risk decision does not bind the supplied portfolio target")
+        target_weights = _immutable_weights(target.weights)
+        payload = {
+            "portfolio_target_id": target.target_id,
+            "risk_decision_id": decision.decision_id,
+            "environment": "paper",
+            "target_weights": target_weights,
+            "created_at": decision.created_at,
+        }
+        object.__setattr__(self, "intent_id", _stable_id("order-intent", payload))
+        object.__setattr__(self, "portfolio_target_id", target.target_id)
+        object.__setattr__(self, "risk_decision_id", decision.decision_id)
+        object.__setattr__(self, "environment", "paper")
+        object.__setattr__(self, "target_weights", target_weights)
+        object.__setattr__(self, "created_at", decision.created_at)
+
+
+@dataclass(frozen=True, slots=True)
+class GovernedFactorSliceResult:
+    dataset_snapshot: DatasetSnapshot
+    factor_version: FactorVersion
+    strategy_version: StrategyVersion
+    research_result: FactorBacktestResult
+    backtest_run: BacktestRun
+    portfolio_target: PortfolioTarget
+    risk_decision: RiskDecision
+    order_intent: PaperOrderIntent | None
+
+
+def evaluate_portfolio_risk(
+    target: PortfolioTarget,
+    policy: RiskPolicy,
+    *,
+    created_at: datetime | None = None,
+) -> RiskDecision:
+    """Apply fail-closed paper risk limits to a versioned target portfolio."""
+    decision_time = target.created_at if created_at is None else _aware_utc(created_at, "created_at")
+    reasons: list[str] = []
+    if target.gross_exposure > policy.max_gross_exposure + 1e-12:
+        reasons.append(
+            f"gross exposure {target.gross_exposure:.12g} exceeds "
+            f"limit {policy.max_gross_exposure:.12g}"
+        )
+    positive_weights = [weight for weight in target.weights.values() if weight > 0]
+    largest_weight = max(positive_weights, default=0.0)
+    if largest_weight > policy.max_single_asset_weight + 1e-12:
+        reasons.append(
+            f"single-asset weight {largest_weight:.12g} exceeds "
+            f"limit {policy.max_single_asset_weight:.12g}"
+        )
+    if len(positive_weights) > policy.max_positions:
+        reasons.append(
+            f"position count {len(positive_weights)} exceeds limit {policy.max_positions}"
+        )
+    status = RiskDecisionStatus.REJECTED if reasons else RiskDecisionStatus.APPROVED
+    payload = {
+        "portfolio_target_id": target.target_id,
+        "policy_id": policy.policy_id,
+        "status": status,
+        "reasons": reasons,
+        "created_at": decision_time,
+    }
+    return RiskDecision(
+        decision_id=_stable_id("risk-decision", payload),
+        portfolio_target_id=target.target_id,
+        policy_id=policy.policy_id,
+        status=status,
+        reasons=tuple(reasons),
+        created_at=decision_time,
+    )
+
+
+def create_paper_order_intent(
+    target: PortfolioTarget,
+    decision: RiskDecision,
+) -> PaperOrderIntent:
+    """Create a paper-only intent after an exact, approved risk decision."""
+    return PaperOrderIntent(target, decision)
+
+
+def run_governed_factor_slice(
+    *,
+    factor_scores: pd.DataFrame,
+    execution_prices: pd.DataFrame,
+    valuation_prices: pd.DataFrame,
+    dataset_snapshot: DatasetSnapshot,
+    factor_version: FactorVersion,
+    strategy_version: StrategyVersion,
+    risk_policy: RiskPolicy,
+    code_revision: str,
+    created_at: datetime,
+    top_k: int,
+    execution_price_field: str,
+    valuation_price_field: str,
+    lag_sessions: int = 1,
+    gross_exposure: float = 1.0,
+    largest: bool = True,
+    initial_cash: float = 1_000_000.0,
+    execution_config: ExecutionConfig | None = None,
+) -> GovernedFactorSliceResult:
+    """Run snapshot → factor → strategy → backtest → target → risk → paper intent."""
+    if strategy_version.factor_version_id != factor_version.version_id:
+        raise ValueError("strategy factor lineage does not match factor_version")
+    if dataset_snapshot.schema_version != factor_version.dataset_schema_version:
+        raise ValueError("factor dataset schema does not match dataset snapshot")
+    if strategy_version.stage not in {StrategyStage.APPROVED, StrategyStage.PAPER}:
+        raise ValueError("strategy must be in Approved or Paper stage")
+    normalized_revision = code_revision.strip().lower()
+    if not _GIT_SHA.fullmatch(normalized_revision):
+        raise ValueError("code_revision must be a lowercase 40-character Git SHA")
+    run_time = _aware_utc(created_at, "created_at")
+    if dataset_snapshot.available_at > run_time or dataset_snapshot.ingested_at > run_time:
+        raise ValueError("dataset snapshot must be available before the research run")
+    if not factor_scores.empty:
+        latest_decision = pd.Timestamp(factor_scores.index.max())
+        latest_decision_date = latest_decision.date()
+        if latest_decision_date > run_time.date():
+            raise ValueError("factor_scores contain future decision dates")
+    config = ExecutionConfig() if execution_config is None else execution_config
+    if not isinstance(config, ExecutionConfig):
+        raise TypeError("execution_config must be an ExecutionConfig")
+
+    parameters: dict[str, object] = {
+        "top_k": top_k,
+        "execution_price_field": execution_price_field,
+        "valuation_price_field": valuation_price_field,
+        "lag_sessions": lag_sessions,
+        "gross_exposure": gross_exposure,
+        "largest": largest,
+        "initial_cash": initial_cash,
+        "execution_config": asdict(config),
+    }
+    config_hash = _stable_id("config", parameters).split(":", maxsplit=1)[1]
+    run_payload = {
+        "dataset_snapshot_id": dataset_snapshot.snapshot_id,
+        "dataset_content_sha256": dataset_snapshot.content_sha256,
+        "factor_version_id": factor_version.version_id,
+        "factor_definition_sha256": factor_version.definition_sha256,
+        "strategy_version_id": strategy_version.version_id,
+        "code_revision": normalized_revision,
+        "config_hash": config_hash,
+        "created_at": run_time,
+    }
+    backtest_run = BacktestRun(
+        run_id=_stable_id("backtest-run", run_payload),
+        dataset_snapshot_id=dataset_snapshot.snapshot_id,
+        factor_version_id=factor_version.version_id,
+        strategy_version_id=strategy_version.version_id,
+        code_revision=normalized_revision,
+        config_hash=config_hash,
+        created_at=run_time,
+    )
+
+    research_result = run_factor_backtest_research(
+        factor_scores,
+        execution_prices,
+        valuation_prices,
+        top_k=top_k,
+        execution_price_field=execution_price_field,
+        valuation_price_field=valuation_price_field,
+        lag_sessions=lag_sessions,
+        gross_exposure=gross_exposure,
+        largest=largest,
+        initial_cash=initial_cash,
+        config=config,
+    )
+    if research_result.schedule.decision_weights.empty:
+        raise ValueError("governed slice requires at least one target portfolio")
+    final_weights = {
+        str(asset): float(weight)
+        for asset, weight in research_result.schedule.decision_weights.iloc[-1].items()
+    }
+    target_payload = {
+        "backtest_run_id": backtest_run.run_id,
+        "dataset_snapshot_id": dataset_snapshot.snapshot_id,
+        "weights": final_weights,
+        "created_at": run_time,
+    }
+    portfolio_target = PortfolioTarget(
+        target_id=_stable_id("portfolio-target", target_payload),
+        backtest_run_id=backtest_run.run_id,
+        dataset_snapshot_id=dataset_snapshot.snapshot_id,
+        weights=final_weights,
+        created_at=run_time,
+    )
+    risk_decision = evaluate_portfolio_risk(
+        portfolio_target,
+        risk_policy,
+        created_at=run_time,
+    )
+    order_intent = (
+        create_paper_order_intent(portfolio_target, risk_decision)
+        if risk_decision.status is RiskDecisionStatus.APPROVED
+        else None
+    )
+    return GovernedFactorSliceResult(
+        dataset_snapshot=dataset_snapshot,
+        factor_version=factor_version,
+        strategy_version=strategy_version,
+        research_result=research_result,
+        backtest_run=backtest_run,
+        portfolio_target=portfolio_target,
+        risk_decision=risk_decision,
+        order_intent=order_intent,
+    )
+
+
+def _demo() -> Mapping[str, object]:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    scores = pd.DataFrame({"A": [2.0, 1.0], "B": [1.0, 2.0]}, index=dates[:2])
+    opens = pd.DataFrame({"A": [10.0, 10.0, 10.2, 10.3], "B": [20.0, 20.0, 20.5, 21.0]}, index=dates)
+    result = run_governed_factor_slice(
+        factor_scores=scores,
+        execution_prices=opens,
+        valuation_prices=opens * 1.01,
+        dataset_snapshot=DatasetSnapshot(
+            snapshot_id="dataset:architecture-smoke-v1",
+            schema_version="1.0.0",
+            content_sha256="a" * 64,
+            effective_at=datetime(2026, 1, 8, 7, tzinfo=UTC),
+            available_at=datetime(2026, 1, 8, 8, tzinfo=UTC),
+            ingested_at=datetime(2026, 1, 8, 8, 5, tzinfo=UTC),
+        ),
+        factor_version=FactorVersion(
+            factor_id="factor:architecture-smoke",
+            version="1.0.0",
+            definition_sha256="b" * 64,
+            dataset_schema_version="1.0.0",
+        ),
+        strategy_version=StrategyVersion(
+            strategy_id="strategy:architecture-smoke",
+            version="1.0.0",
+            factor_version_id="factor:architecture-smoke@1.0.0",
+            stage=StrategyStage.APPROVED,
+        ),
+        risk_policy=RiskPolicy(
+            policy_id="risk:architecture-smoke@1.0.0",
+            max_gross_exposure=1.0,
+            max_single_asset_weight=0.6,
+            max_positions=10,
+        ),
+        code_revision="c" * 40,
+        created_at=datetime(2026, 1, 9, 1, tzinfo=UTC),
+        top_k=2,
+        execution_price_field="open",
+        valuation_price_field="close",
+        execution_config=ExecutionConfig(
+            commission_bps=0,
+            stamp_tax_bps=0,
+            slippage_bps=0,
+            min_trade_amount=0,
+        ),
+    )
+    return {
+        "backtest_run_id": result.backtest_run.run_id,
+        "portfolio_target_id": result.portfolio_target.target_id,
+        "risk_decision": result.risk_decision.status.value,
+        "order_intent_id": result.order_intent.intent_id if result.order_intent else None,
+        "environment": result.order_intent.environment if result.order_intent else None,
+    }
+
+
+if __name__ == "__main__":
+    print(json.dumps(_demo(), ensure_ascii=False, sort_keys=True))

tests/fixtures/factor-contracts-v1.golden.json

@@ -0,0 +1,206 @@
+{
+  "dataset_snapshot": {
+    "contract_name": "researchhub.dataset-snapshot",
+    "schema_version": "1.0.0",
+    "snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57",
+    "descriptor": {
+      "dataset": {
+        "dataset_id": "rhdataset:market:0123456789abcdef0123456789abcdef",
+        "dataset_kind": "market",
+        "record_schema_version": "1.0.0",
+        "dimensions": ["instrument_id", "effective_time"]
+      },
+      "published_at": "2026-01-02T07:05:00Z",
+      "time_semantics": {
+        "effective_time": {
+          "start_inclusive": "2026-01-02T07:00:00Z",
+          "end_inclusive": "2026-01-02T07:00:00Z"
+        },
+        "knowledge_time": {
+          "start_inclusive": "2026-01-02T07:01:00Z",
+          "end_inclusive": "2026-01-02T07:01:00Z"
+        },
+        "pit_cutoff": "2026-01-02T07:01:00Z"
+      },
+      "content": {
+        "digest_algorithm": "sha256",
+        "canonicalization": "RFC8785",
+        "record_order": "canonical-record-byte-order",
+        "content_digest": "sha256:44ea11ba64dc2e6fd55c6d8e038c5edc84ee38d6fd46e38b15a1d5409662a020",
+        "logical_manifest": {
+          "record_count": 2,
+          "chunks": [
+            {
+              "chunk_index": 0,
+              "content_digest": "sha256:44ea11ba64dc2e6fd55c6d8e038c5edc84ee38d6fd46e38b15a1d5409662a020",
+              "record_count": 2
+            }
+          ]
+        },
+        "manifest_digest": "sha256:d991bb2f8f6b80525f93c51e0b371213a3ed4649dffb073ed4605bfbd32349bd",
+        "record_count": 2
+      },
+      "lineage": {
+        "publisher": {"id": "researchhub.data", "version": "1.0.0"},
+        "transformation": {
+          "id": "rhtransform:00112233445566778899aabbccddeeff",
+          "version": "1.0.0"
+        },
+        "upstream_snapshot_ids": [],
+        "upstream_content_digests": []
+      },
+      "quality": {
+        "status": "passed",
+        "checks": [
+          {
+            "check_id": "completeness",
+            "status": "passed",
+            "severity": "blocking",
+            "evidence_digest": "sha256:876fc2fcc6414ddc3f824a47f475d34c82a53d2bda5dc72a234a3f3164e8e2ec"
+          },
+          {
+            "check_id": "pit_time_integrity",
+            "status": "passed",
+            "severity": "blocking",
+            "evidence_digest": "sha256:90a6cc46b9f2ab317a1c6d14dc173784e19b5621cd7fe956e8f41338cbdc5944"
+          }
+        ]
+      },
+      "qualification": {
+        "status": "qualified",
+        "policy_id": "researchhub.dataset-snapshot.pit",
+        "policy_version": "1.0.0",
+        "evaluated_at": "2026-01-02T07:04:00Z",
+        "evidence_digest": "sha256:e192462f9022f2b477f73cdbe9e6c9f891ebcfdc2b4ed4f8ddd7b1ff107ee6a6"
+      }
+    }
+  },
+  "data_foundation": {
+    "contract_name": "researchhub.data-foundation",
+    "schema_version": "1.0.0",
+    "foundation_id": "rhdfv1:sha256:d848237ab753ee9432ae78ec1f93b6ac45c8072d6694023b7f288203daf9d838",
+    "dataset_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57",
+    "pit_cutoff": "2026-01-03T00:00:00Z",
+    "instrument_routes": [
+      {
+        "route_revision_id": "rhroutev1:sha256:ca67013250e28ab4cce16570607379a8792400e62415ee6cb71c75508e2f3d86",
+        "instrument_id": "rhinstrument:0123456789abcdef0123456789abcdef",
+        "revision_number": 1,
+        "symbol": "600000",
+        "mic": "XSHG",
+        "currency": "CNY",
+        "asset_class": "equity",
+        "instrument_type": "stock",
+        "calendar_id": "rhcalendar:11112222333344445555666677778888",
+        "effective_from": "2020-01-01T00:00:00Z",
+        "knowledge_time": "2026-01-01T07:00:00Z",
+        "evidence_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+      }
+    ],
+    "trading_calendar_revisions": [
+      {
+        "calendar_revision_id": "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078",
+        "calendar_id": "rhcalendar:11112222333344445555666677778888",
+        "session_date": "2026-01-02",
+        "revision_number": 1,
+        "status": "open",
+        "sessions": [
+          {"opens_at": "2026-01-02T01:30:00Z", "closes_at": "2026-01-02T07:00:00Z"}
+        ],
+        "knowledge_time": "2026-01-01T08:00:00Z",
+        "evidence_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+      }
+    ],
+    "corporate_action_revisions": [
+      {
+        "action_revision_id": "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275",
+        "action_id": "rhaction:99998888777766665555444433332222",
+        "instrument_id": "rhinstrument:0123456789abcdef0123456789abcdef",
+        "revision_number": 1,
+        "action_type": "cash_dividend",
+        "status": "confirmed",
+        "effective_time": "2026-01-02T00:00:00Z",
+        "knowledge_time": "2026-01-01T09:00:00Z",
+        "terms_digest": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+        "evidence_digest": "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
+      }
+    ],
+    "standardized_views": [
+      {
+        "view_ref_id": "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c",
+        "view_id": "rhview:abcdef0123456789abcdef0123456789",
+        "view_version": "1.0.0",
+        "dataset_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57",
+        "pit_cutoff": "2026-01-03T00:00:00Z",
+        "schema_digest": "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+        "content_digest": "sha256:123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0",
+        "transformation_digest": "sha256:23456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef01",
+        "instrument_route_revision_ids": [
+          "rhroutev1:sha256:ca67013250e28ab4cce16570607379a8792400e62415ee6cb71c75508e2f3d86"
+        ],
+        "trading_calendar_revision_ids": [
+          "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078"
+        ],
+        "corporate_action_revision_ids": [
+          "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275"
+        ]
+      }
+    ],
+    "revision_lineage": [
+      {
+        "revision_kind": "instrument_route",
+        "revision_id": "rhroutev1:sha256:ca67013250e28ab4cce16570607379a8792400e62415ee6cb71c75508e2f3d86",
+        "revision_number": 1,
+        "knowledge_time": "2026-01-01T07:00:00Z",
+        "evidence_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+      },
+      {
+        "revision_kind": "trading_calendar",
+        "revision_id": "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078",
+        "revision_number": 1,
+        "knowledge_time": "2026-01-01T08:00:00Z",
+        "evidence_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+      },
+      {
+        "revision_kind": "corporate_action",
+        "revision_id": "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275",
+        "revision_number": 1,
+        "knowledge_time": "2026-01-01T09:00:00Z",
+        "evidence_digest": "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
+      }
+    ],
+    "readiness": {
+      "evidence_scope": "synthetic_fixture",
+      "contract_validation": {
+        "status": "validated",
+        "evidence_digests": [
+          "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
+        ]
+      },
+      "real_data_validation": {"status": "not_validated", "evidence_digests": []},
+      "production_validation": {"status": "not_validated", "evidence_digests": []},
+      "live_validation": {"status": "not_validated", "evidence_digests": []}
+    }
+  },
+  "output_schema": {
+    "columns": ["evaluation_at", "factor_id", "instrument_id", "value"],
+    "schema_version": "1.0.0"
+  },
+  "output_content": {
+    "rows": [
+      {
+        "evaluation_at": "2026-01-03T11:00:00Z",
+        "factor_id": "alpha_005",
+        "instrument_id": "rhinstrument:0123456789abcdef0123456789abcdef",
+        "value": "0.125"
+      }
+    ]
+  },
+  "expected": {
+    "definition_id": "rhfactorv1:sha256:978fb8000d318373844a5e044ca14bf377e01ebe8d85964b826ecd2af9085ce9",
+    "input_schema_digest": "sha256:4501aeab99b4bcc25a1b8813ebe197fb498053fd710d73746bf20fc8eeb4bfa7",
+    "factor_set_id": "rhfactorsetv1:sha256:e9339581cf569e92459f672e8081337712e7bf98e58ad42d60d7ed13f9b5a021",
+    "output_artifact_id": "rhfactoroutputv1:sha256:a4803b5ff66d12d3a0e7e8e5b8cca953cbc137e2bf41514b7c4e5f05da5ee68b",
+    "legacy_binding_id": "rhlegacyfactorv1:sha256:541bc5a9469f9c8e4c2d696a9972fc5f2e6e2bef218b86f728823994b915dede"
+  }
+}

tests/governance/test_module_spec.py

@@ -1,32 +1,51 @@
 from __future__ import annotations
 
 import json
-import unittest
 from pathlib import Path
 
 
 ROOT = Path(__file__).resolve().parents[2]
 
 
-class ModuleSpecTests(unittest.TestCase):
-    def test_module_spec_declares_pure_research_engine_boundary(self) -> None:
-        spec = json.loads((ROOT / "MODULE_SPEC.yaml").read_text(encoding="utf-8"))
-        self.assertEqual(spec["module_id"], "quant_engine")
-        self.assertEqual(spec["authority"]["subject"], spec["module_id"])
-        self.assertEqual(spec["repository"]["type"], "research_engine")
-        self.assertEqual(spec["bounded_context"]["domain"], "quantitative-research-engine")
-        prohibited = " ".join(spec["bounded_context"]["prohibited_responsibilities"]).lower()
-        for term in ("investment advice", "live order", "credentials", "source facts"):
-            self.assertIn(term, prohibited)
-        self.assertEqual(spec["contracts"], {"provides": [], "consumes": []})
-        self.assertEqual(spec["dependencies"], [])
-        self.assertTrue(
-            all(
-                command["required"] and not command["network"]
-                for command in spec["verification"]["commands"]
-            )
-        )
+def test_module_spec_declares_pure_research_engine_boundary() -> None:
+    spec = json.loads((ROOT / "MODULE_SPEC.yaml").read_text(encoding="utf-8"))
+    assert spec["module_id"] == "quant_engine"
+    assert spec["authority"]["subject"] == spec["module_id"]
+    assert spec["repository"]["type"] == "research_engine"
+    assert spec["bounded_context"]["domain"] == "quantitative-research-engine"
+    prohibited = " ".join(spec["bounded_context"]["prohibited_responsibilities"]).lower()
+    for term in ("investment advice", "live order", "credentials", "source facts"):
+        assert term in prohibited
+    assert spec["authority"]["revision"] == 2
+    assert {
+        (item["contract_id"], item["version"])
+        for item in spec["contracts"]["provides"]
+    } == {
+        ("researchhub.factor-definition", "1.0.0"),
+        ("researchhub.factor-set-ref", "1.0.0"),
+    }
+    assert all(
+        item["authority"] == "quant_engine"
+        and item["path"] == "src/quant_engine/factor_contracts.py"
+        for item in spec["contracts"]["provides"]
+    )
+    assert {
+        (item["contract_id"], item["version"])
+        for item in spec["contracts"]["consumes"]
+    } == {
+        ("researchhub.dataset-snapshot", "1.0.0"),
+        ("researchhub.data-foundation", "1.0.0"),
+    }
+    assert all(
+        item["authority"] == "researchhub.data"
+        for item in spec["contracts"]["consumes"]
+    )
+    assert spec["dependencies"] == []
+    assert all(
+        command["required"] and not command["network"]
+        for command in spec["verification"]["commands"]
+    )
 
 
 if __name__ == "__main__":
-    unittest.main()
+    test_module_spec_declares_pure_research_engine_boundary()

tests/test_alpha_factors.py

@@ -7,6 +7,13 @@ import pandas as pd
 import pytest
 
 import quant_engine.alpha_factors as alpha_factors_module
+from quant_engine.factor_contracts import (
+    FactorContractError,
+    FactorInput,
+    ProducerIdentity,
+    factor_definition_from_alpha158,
+    factor_input_schema_digest,
+)
 from quant_engine.alpha_factors import (
     ALPHA158_REGISTRY,
     ALPHA158_PHASE1_OPERATOR_SPECS,
@@ -14,6 +21,15 @@ from quant_engine.alpha_factors import (
     ALPHA158_PHASE3_FORMULA_CATALOG_SHA256,
     ALPHA158_PHASE3_FORMULA_CONTRACT_VERSION,
     ALPHA158_PHASE3_FORMULA_SPECS,
+    ALPHA158_PHASE4_FORMULA_CATALOG_SHA256,
+    ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION,
+    ALPHA158_PHASE4_FORMULA_SPECS,
+    ALPHA158_PHASE5_FORMULA_CATALOG_SHA256,
+    ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION,
+    ALPHA158_PHASE5_FORMULA_SPECS,
+    ALPHA158_PHASE6_FORMULA_CATALOG_SHA256,
+    ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION,
+    ALPHA158_PHASE6_FORMULA_SPECS,
     alpha_001,
     alpha_002,
     alpha_003,
@@ -175,9 +191,15 @@ from quant_engine.alpha_factors import (
     evaluate_phase1_operator,
     evaluate_phase2_operator,
     evaluate_phase3_formula,
+    evaluate_phase4_formula,
+    evaluate_phase5_formula,
+    evaluate_phase6_formula,
     list_phase1_operators,
     list_phase2_operators,
     list_phase3_formulas,
+    list_phase4_formulas,
+    list_phase5_formulas,
+    list_phase6_formulas,
     correlation,
     covariance,
     decay_linear,
@@ -419,6 +441,50 @@ def test_alpha_registry_required_fields():
         assert required <= set(meta.keys()), f"{alpha_id} missing fields"
 
 
+def test_alpha_registry_adapts_to_definition_without_copying_formula_or_inputs():
+    factor_input = FactorInput(
+        "market",
+        "sha256:" + "1" * 64,
+        tuple(ALPHA158_REGISTRY["alpha_005"]["inputs"]),
+    )
+    definition = factor_definition_from_alpha158(
+        "alpha_005",
+        version="1.0.0",
+        parameters={},
+        inputs=(factor_input,),
+        implementation_digest="sha256:" + "2" * 64,
+        input_schema_digest=factor_input_schema_digest((factor_input,)),
+        valid_from="2026-01-01T00:00:00Z",
+        valid_until="2027-01-01T00:00:00Z",
+        warmup_sessions=10,
+        lag_sessions=1,
+        producer=ProducerIdentity("quant_engine", "1.0.0"),
+        code_revision="c" * 40,
+    )
+
+    assert definition.formula == ALPHA158_REGISTRY["alpha_005"]["formula"]
+    assert definition.inputs[0].required_columns == tuple(
+        ALPHA158_REGISTRY["alpha_005"]["inputs"]
+    )
+
+    incomplete = FactorInput("market", "sha256:" + "1" * 64, ("close",))
+    with pytest.raises(FactorContractError, match="exactly correspond"):
+        factor_definition_from_alpha158(
+            "alpha_005",
+            version="1.0.0",
+            parameters={},
+            inputs=(incomplete,),
+            implementation_digest="sha256:" + "2" * 64,
+            input_schema_digest=factor_input_schema_digest((incomplete,)),
+            valid_from="2026-01-01T00:00:00Z",
+            valid_until="2027-01-01T00:00:00Z",
+            warmup_sessions=10,
+            lag_sessions=1,
+            producer=ProducerIdentity("quant_engine", "1.0.0"),
+            code_revision="c" * 40,
+        )
+
+
 def test_get_alpha_meta_success():
     """已知 alpha_id 返回完整 meta。"""
     meta = get_alpha_meta("alpha_001")
@@ -1641,3 +1707,504 @@ def test_phase3_dispatch_rejects_implicit_series_alignment():
             close=inputs["close"],
             volume=misaligned_volume,
         )
+
+
+# ── Alpha158 Phase 4: versioned alpha051-alpha100 formula contract ──────────
+
+
+def test_phase4_formula_catalog_is_versioned_exact_and_content_addressed():
+    import hashlib
+    import json
+    from collections import Counter
+
+    expected_ids = tuple(f"alpha_{number:03d}" for number in range(51, 101))
+    expected_fields = {
+        "name",
+        "contract_version",
+        "formula",
+        "category",
+        "complexity",
+        "parameters",
+        "description",
+        "references",
+        "call_inputs",
+        "formula_inputs",
+        "input_category",
+    }
+
+    assert ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION == "1.0.0"
+    assert list_phase4_formulas() == expected_ids
+    assert tuple(ALPHA158_PHASE4_FORMULA_SPECS) == expected_ids
+    assert Counter(
+        spec["input_category"] for spec in ALPHA158_PHASE4_FORMULA_SPECS.values()
+    ) == {"single": 1, "pair": 27, "triple": 11, "quadruple": 10, "quintuple": 1}
+
+    for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items():
+        assert set(spec) == expected_fields
+        assert spec["name"] == alpha_id
+        assert spec["contract_version"] == ALPHA158_PHASE4_FORMULA_CONTRACT_VERSION
+        assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
+
+    serializable_specs = {
+        alpha_id: {
+            field: list(value) if isinstance(value, tuple) else value
+            for field, value in spec.items()
+        }
+        for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items()
+    }
+    encoded = json.dumps(
+        serializable_specs,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode()
+    assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE4_FORMULA_CATALOG_SHA256
+    assert ALPHA158_PHASE4_FORMULA_CATALOG_SHA256 == (
+        "858daf5e2abab5063fc28fcf7c79936096e2c76dde582fc7bab78b3458f17054"
+    )
+
+
+def test_phase4_formula_catalog_is_recursively_immutable():
+    import operator
+
+    with pytest.raises(TypeError):
+        operator.setitem(ALPHA158_PHASE4_FORMULA_SPECS, "alpha_051", {})
+    with pytest.raises(TypeError):
+        operator.setitem(
+            ALPHA158_PHASE4_FORMULA_SPECS["alpha_051"],
+            "formula",
+            "changed",
+        )
+    with pytest.raises(TypeError):
+        operator.setitem(
+            ALPHA158_PHASE4_FORMULA_SPECS["alpha_051"]["call_inputs"],
+            0,
+            "volume",
+        )
+
+
+def test_phase4_catalog_freezes_callable_and_formula_inputs():
+    import inspect
+
+    for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items():
+        function = getattr(alpha_factors_module, alpha_id)
+        signature_inputs = tuple(
+            "open" if name == "open_" else name
+            for name in inspect.signature(function).parameters
+        )
+        assert spec["call_inputs"] == signature_inputs
+        assert spec["formula_inputs"] == tuple(ALPHA158_REGISTRY[alpha_id]["inputs"])
+
+    assert ALPHA158_PHASE4_FORMULA_SPECS["alpha_055"]["call_inputs"] == (
+        "open",
+        "high",
+        "low",
+        "volume",
+        "close",
+    )
+
+
+def test_phase4_dispatch_matches_all_existing_alpha051_alpha100_functions():
+    inputs = _phase3_market_inputs()
+
+    for alpha_id, spec in ALPHA158_PHASE4_FORMULA_SPECS.items():
+        call_inputs = spec["call_inputs"]
+        function = getattr(alpha_factors_module, alpha_id)
+        expected = function(*(inputs[name] for name in call_inputs))
+        actual = evaluate_phase4_formula(
+            alpha_id,
+            **{name: inputs[name] for name in reversed(call_inputs)},
+        )
+        pd.testing.assert_series_equal(actual, expected)
+
+
+def test_phase4_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
+    inputs = _phase3_market_inputs()
+
+    with pytest.raises(KeyError, match="not registered"):
+        evaluate_phase4_formula("alpha_050", close=inputs["close"])
+    with pytest.raises(ValueError, match=r"missing inputs.*low"):
+        evaluate_phase4_formula("alpha_051", high=inputs["high"])
+    with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
+        evaluate_phase4_formula(
+            "alpha_051",
+            high=inputs["high"],
+            low=inputs["low"],
+            vwap=inputs["vwap"],
+        )
+    with pytest.raises(TypeError, match="high must be a pandas Series"):
+        evaluate_phase4_formula(  # type: ignore[arg-type]
+            "alpha_051",
+            high=[1.0, 2.0],
+            low=inputs["low"],
+        )
+
+
+def test_phase4_dispatch_rejects_implicit_series_alignment():
+    inputs = _phase3_market_inputs()
+    misaligned_low = inputs["low"].rename(index={79: 80})
+
+    with pytest.raises(ValueError, match="low index must align with high"):
+        evaluate_phase4_formula(
+            "alpha_051",
+            high=inputs["high"],
+            low=misaligned_low,
+        )
+
+
+# ── Alpha158 Phase 5: versioned alpha101-alpha150 formula contract ──────────
+
+
+def test_phase5_formula_catalog_is_versioned_exact_and_content_addressed():
+    import hashlib
+    import json
+    from collections import Counter
+
+    expected_ids = tuple(f"alpha_{number:03d}" for number in range(101, 151))
+    expected_fields = {
+        "name",
+        "contract_version",
+        "formula",
+        "category",
+        "complexity",
+        "parameters",
+        "description",
+        "references",
+        "call_inputs",
+        "formula_inputs",
+        "input_category",
+    }
+
+    assert ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION == "1.0.0"
+    assert list_phase5_formulas() == expected_ids
+    assert tuple(ALPHA158_PHASE5_FORMULA_SPECS) == expected_ids
+    assert Counter(
+        spec["input_category"] for spec in ALPHA158_PHASE5_FORMULA_SPECS.values()
+    ) == {"pair": 33, "triple": 14, "quadruple": 3}
+
+    for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items():
+        assert set(spec) == expected_fields
+        assert spec["name"] == alpha_id
+        assert spec["contract_version"] == ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION
+        assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
+
+    serializable_specs = {
+        alpha_id: {
+            field: list(value) if isinstance(value, tuple) else value
+            for field, value in spec.items()
+        }
+        for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items()
+    }
+    encoded = json.dumps(
+        serializable_specs,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode()
+    assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE5_FORMULA_CATALOG_SHA256
+    assert ALPHA158_PHASE5_FORMULA_CATALOG_SHA256 == (
+        "3368796169c9fbd39c4a34ea137e569964b15882fbf4de25124790d548db6533"
+    )
+
+
+def test_phase5_formula_catalog_is_recursively_immutable():
+    import operator
+
+    with pytest.raises(TypeError):
+        operator.setitem(ALPHA158_PHASE5_FORMULA_SPECS, "alpha_101", {})
+    with pytest.raises(TypeError):
+        operator.setitem(
+            ALPHA158_PHASE5_FORMULA_SPECS["alpha_101"],
+            "formula",
+            "changed",
+        )
+    with pytest.raises(TypeError):
+        operator.setitem(
+            ALPHA158_PHASE5_FORMULA_SPECS["alpha_101"]["call_inputs"],
+            0,
+            "volume",
+        )
+
+
+def test_phase5_catalog_freezes_callable_and_formula_inputs():
+    import inspect
+
+    for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items():
+        function = getattr(alpha_factors_module, alpha_id)
+        signature_inputs = tuple(
+            "open" if name == "open_" else name
+            for name in inspect.signature(function).parameters
+        )
+        assert spec["call_inputs"] == signature_inputs
+        assert spec["formula_inputs"] == tuple(ALPHA158_REGISTRY[alpha_id]["inputs"])
+
+    assert ALPHA158_PHASE5_FORMULA_SPECS["alpha_101"]["call_inputs"] == (
+        "close",
+        "high",
+        "low",
+    )
+
+
+def test_phase5_dispatch_matches_all_existing_alpha101_alpha150_functions():
+    inputs = _phase3_market_inputs()
+
+    for alpha_id, spec in ALPHA158_PHASE5_FORMULA_SPECS.items():
+        call_inputs = spec["call_inputs"]
+        function = getattr(alpha_factors_module, alpha_id)
+        expected = function(*(inputs[name] for name in call_inputs))
+        actual = evaluate_phase5_formula(
+            alpha_id,
+            **{name: inputs[name] for name in reversed(call_inputs)},
+        )
+        pd.testing.assert_series_equal(actual, expected)
+
+
+def test_phase5_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
+    inputs = _phase3_market_inputs()
+
+    with pytest.raises(KeyError, match="not registered"):
+        evaluate_phase5_formula("alpha_100", close=inputs["close"])
+    with pytest.raises(KeyError, match="not registered"):
+        evaluate_phase5_formula("alpha_151", close=inputs["close"])
+    with pytest.raises(ValueError, match=r"missing inputs.*low"):
+        evaluate_phase5_formula(
+            "alpha_101",
+            close=inputs["close"],
+            high=inputs["high"],
+        )
+    with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
+        evaluate_phase5_formula(
+            "alpha_101",
+            close=inputs["close"],
+            high=inputs["high"],
+            low=inputs["low"],
+            vwap=inputs["vwap"],
+        )
+    with pytest.raises(TypeError, match="high must be a pandas Series"):
+        evaluate_phase5_formula(  # type: ignore[arg-type]
+            "alpha_101",
+            close=inputs["close"],
+            high=[1.0, 2.0],
+            low=inputs["low"],
+        )
+
+
+def test_phase5_dispatch_rejects_length_and_index_alignment_errors():
+    inputs = _phase3_market_inputs()
+    shorter_low = inputs["low"].iloc[:-1]
+    misaligned_high = inputs["high"].rename(index={79: 80})
+
+    with pytest.raises(ValueError, match="low length must match close"):
+        evaluate_phase5_formula(
+            "alpha_101",
+            close=inputs["close"],
+            high=inputs["high"],
+            low=shorter_low,
+        )
+    with pytest.raises(ValueError, match="high index must align with close"):
+        evaluate_phase5_formula(
+            "alpha_101",
+            close=inputs["close"],
+            high=misaligned_high,
+            low=inputs["low"],
+        )
+
+
+def test_phase5_contract_is_publicly_exported():
+    assert {
+        "ALPHA158_PHASE5_FORMULA_CONTRACT_VERSION",
+        "ALPHA158_PHASE5_FORMULA_CATALOG_SHA256",
+        "ALPHA158_PHASE5_FORMULA_SPECS",
+        "list_phase5_formulas",
+        "evaluate_phase5_formula",
+    } <= set(alpha_factors_module.__all__)
+
+
+# ── Alpha158 Phase 6: versioned alpha151-alpha158 formula contract ──────────
+
+
+def test_phase6_formula_catalog_is_versioned_exact_and_content_addressed():
+    import hashlib
+    import json
+    from collections import Counter
+
+    expected_ids = tuple(f"alpha_{number:03d}" for number in range(151, 159))
+    expected_fields = {
+        "name",
+        "contract_version",
+        "formula",
+        "category",
+        "complexity",
+        "parameters",
+        "description",
+        "references",
+        "call_inputs",
+        "formula_inputs",
+        "input_category",
+    }
+
+    assert ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION == "1.0.0"
+    assert list_phase6_formulas() == expected_ids
+    assert tuple(ALPHA158_PHASE6_FORMULA_SPECS) == expected_ids
+    assert Counter(
+        spec["input_category"] for spec in ALPHA158_PHASE6_FORMULA_SPECS.values()
+    ) == {"pair": 6, "triple": 2}
+
+    for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items():
+        assert set(spec) == expected_fields
+        assert spec["name"] == alpha_id
+        assert spec["contract_version"] == ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION
+        assert spec["formula"] == ALPHA158_REGISTRY[alpha_id]["formula"]
+
+    serializable_specs = {
+        alpha_id: {
+            field: list(value) if isinstance(value, tuple) else value
+            for field, value in spec.items()
+        }
+        for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items()
+    }
+    encoded = json.dumps(
+        serializable_specs,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode()
+    assert hashlib.sha256(encoded).hexdigest() == ALPHA158_PHASE6_FORMULA_CATALOG_SHA256
+    assert ALPHA158_PHASE6_FORMULA_CATALOG_SHA256 == (
+        "70ffae16ca6cbb59a1e8644f5cdeec1d291fa75ff8b5647fb534af0083408219"
+    )
+
+
+def test_phase6_formula_catalog_is_recursively_immutable():
+    import operator
+
+    with pytest.raises(TypeError):
+        operator.setitem(ALPHA158_PHASE6_FORMULA_SPECS, "alpha_151", {})
+    with pytest.raises(TypeError):
+        operator.setitem(
+            ALPHA158_PHASE6_FORMULA_SPECS["alpha_151"],
+            "formula",
+            "changed",
+        )
+    with pytest.raises(TypeError):
+        operator.setitem(
+            ALPHA158_PHASE6_FORMULA_SPECS["alpha_151"]["call_inputs"],
+            0,
+            "volume",
+        )
+
+
+def test_phase6_catalog_freezes_callable_and_formula_inputs():
+    import inspect
+
+    for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items():
+        function = getattr(alpha_factors_module, alpha_id)
+        signature_inputs = tuple(
+            "open" if name == "open_" else name
+            for name in inspect.signature(function).parameters
+        )
+        assert spec["call_inputs"] == signature_inputs
+        assert spec["formula_inputs"] == tuple(ALPHA158_REGISTRY[alpha_id]["inputs"])
+
+    assert ALPHA158_PHASE6_FORMULA_SPECS["alpha_158"]["call_inputs"] == (
+        "high",
+        "low",
+        "volume",
+    )
+
+
+def test_phase6_dispatch_matches_all_existing_alpha151_alpha158_functions():
+    inputs = _phase3_market_inputs()
+
+    for alpha_id, spec in ALPHA158_PHASE6_FORMULA_SPECS.items():
+        call_inputs = spec["call_inputs"]
+        function = getattr(alpha_factors_module, alpha_id)
+        expected = function(*(inputs[name] for name in call_inputs))
+        actual = evaluate_phase6_formula(
+            alpha_id,
+            **{name: inputs[name] for name in reversed(call_inputs)},
+        )
+        pd.testing.assert_series_equal(actual, expected)
+
+
+def test_phase6_dispatch_rejects_unknown_missing_extra_and_non_series_inputs():
+    inputs = _phase3_market_inputs()
+
+    with pytest.raises(KeyError, match="not registered"):
+        evaluate_phase6_formula("alpha_150", close=inputs["close"])
+    with pytest.raises(KeyError, match="not registered"):
+        evaluate_phase6_formula("alpha_159", close=inputs["close"])
+    with pytest.raises(ValueError, match=r"missing inputs.*volume"):
+        evaluate_phase6_formula("alpha_151", close=inputs["close"])
+    with pytest.raises(ValueError, match=r"unexpected inputs.*vwap"):
+        evaluate_phase6_formula(
+            "alpha_151",
+            close=inputs["close"],
+            volume=inputs["volume"],
+            vwap=inputs["vwap"],
+        )
+    with pytest.raises(TypeError, match="volume must be a pandas Series"):
+        evaluate_phase6_formula(  # type: ignore[arg-type]
+            "alpha_151",
+            close=inputs["close"],
+            volume=[1.0, 2.0],
+        )
+
+
+def test_phase6_dispatch_rejects_length_and_index_alignment_errors():
+    inputs = _phase3_market_inputs()
+    shorter_volume = inputs["volume"].iloc[:-1]
+    misaligned_low = inputs["low"].rename(index={79: 80})
+
+    with pytest.raises(ValueError, match="volume length must match close"):
+        evaluate_phase6_formula(
+            "alpha_151",
+            close=inputs["close"],
+            volume=shorter_volume,
+        )
+    with pytest.raises(ValueError, match="low index must align with high"):
+        evaluate_phase6_formula(
+            "alpha_158",
+            high=inputs["high"],
+            low=misaligned_low,
+            volume=inputs["volume"],
+        )
+
+
+def test_phase6_preserves_complete_alpha001_alpha158_formula_body_fingerprint():
+    import ast
+    import hashlib
+    import inspect
+    import json
+    import textwrap
+
+    fingerprints = {}
+    for number in range(1, 159):
+        alpha_id = f"alpha_{number:03d}"
+        source = textwrap.dedent(inspect.getsource(getattr(alpha_factors_module, alpha_id)))
+        node = ast.parse(source).body[0]
+        assert isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        body = ast.dump(
+            ast.Module(body=node.body, type_ignores=[]),
+            include_attributes=False,
+        )
+        fingerprints[alpha_id] = hashlib.sha256(body.encode()).hexdigest()
+
+    encoded = json.dumps(
+        fingerprints,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode()
+    assert hashlib.sha256(encoded).hexdigest() == (
+        "f3ae807983cf8ca083d0b924c3807ffd84a62a5bd354f9efb1a1a16ca40c7da6"
+    )
+
+
+def test_phase6_contract_is_publicly_exported():
+    assert {
+        "ALPHA158_PHASE6_FORMULA_CONTRACT_VERSION",
+        "ALPHA158_PHASE6_FORMULA_CATALOG_SHA256",
+        "ALPHA158_PHASE6_FORMULA_SPECS",
+        "list_phase6_formulas",
+        "evaluate_phase6_formula",
+    } <= set(alpha_factors_module.__all__)

tests/test_factor_contracts.py

@@ -0,0 +1,938 @@
+"""Versioned factor-definition and factor-set contract conformance."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from dataclasses import FrozenInstanceError
+from pathlib import Path
+from typing import Any, Callable
+
+import pytest
+
+from quant_engine.factor_contracts import (
+    ActorIdentity,
+    AvailabilityMode,
+    ContractErrorCode,
+    Causation,
+    DataFoundationEnvelope,
+    DatasetSnapshotEnvelope,
+    FactorContractError,
+    FactorDefinition,
+    FactorInput,
+    FactorSetRef,
+    HistoricalAvailability,
+    InputBinding,
+    LegacyFactorBinding,
+    OutputArtifactRef,
+    OutputCoverage,
+    OutputQuality,
+    OutputQualityCheck,
+    PayloadValidation,
+    ProducerIdentity,
+    TypedParameter,
+    ViewAvailability,
+    canonical_json_bytes,
+    factor_definition_from_alpha158,
+    factor_input_schema_digest,
+    validate_factor_catalog,
+)
+from quant_engine.governed_pipeline import (
+    FactorVersion,
+    bind_legacy_factor,
+    project_legacy_factor,
+)
+
+
+FIXTURE_PATH = Path(__file__).parent / "fixtures" / "factor-contracts-v1.golden.json"
+VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c"
+VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+
+
+def _golden() -> dict[str, Any]:
+    loaded = json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
+    assert isinstance(loaded, dict)
+    return loaded
+
+
+def _sha256(value: bytes) -> str:
+    return f"sha256:{hashlib.sha256(value).hexdigest()}"
+
+
+def _reidentify(item: dict[str, Any], field: str, prefix: str) -> None:
+    payload = {key: value for key, value in item.items() if key != field}
+    item[field] = f"{prefix}{hashlib.sha256(canonical_json_bytes(payload)).hexdigest()}"
+
+
+def _snapshot_and_foundation(
+    fixture: dict[str, Any] | None = None,
+) -> tuple[DatasetSnapshotEnvelope, DataFoundationEnvelope]:
+    source = _golden() if fixture is None else fixture
+    return (
+        DatasetSnapshotEnvelope.from_dict(source["dataset_snapshot"]),
+        DataFoundationEnvelope.from_dict(source["data_foundation"]),
+    )
+
+
+def _definition(
+    *,
+    inputs: tuple[FactorInput, ...] | None = None,
+    **overrides: Any,
+) -> FactorDefinition:
+    factor_inputs = inputs or (
+        FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume")),
+    )
+    arguments: dict[str, Any] = {
+        "factor_id": "alpha_005",
+        "version": "1.0.0",
+        "formula": "correlation(close, volume, 10)",
+        "parameters": {},
+        "implementation_digest": "sha256:" + "1" * 64,
+        "input_schema_digest": factor_input_schema_digest(factor_inputs),
+        "inputs": factor_inputs,
+        "valid_from": "2026-01-01T00:00:00.000000Z",
+        "valid_until": "2027-01-01T00:00:00Z",
+        "warmup_sessions": 10,
+        "lag_sessions": 1,
+        "producer": ProducerIdentity("quant_engine", "1.0.0"),
+        "code_revision": "c" * 40,
+    }
+    arguments.update(overrides)
+    return FactorDefinition.create(**arguments)
+
+
+def _golden_definition() -> FactorDefinition:
+    factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume"))
+    return factor_definition_from_alpha158(
+        "alpha_005",
+        version="1.0.0",
+        parameters={},
+        inputs=(factor_input,),
+        implementation_digest="sha256:" + "1" * 64,
+        input_schema_digest=factor_input_schema_digest((factor_input,)),
+        valid_from="2026-01-01T00:00:00.000000Z",
+        valid_until="2027-01-01T00:00:00Z",
+        warmup_sessions=10,
+        lag_sessions=1,
+        producer=ProducerIdentity("quant_engine", "1.0.0"),
+        code_revision="c" * 40,
+    )
+
+
+def _factor_set_arguments(
+    *,
+    fixture: dict[str, Any] | None = None,
+    snapshot: DatasetSnapshotEnvelope | None = None,
+    foundation: DataFoundationEnvelope | None = None,
+    definition: FactorDefinition | None = None,
+) -> dict[str, Any]:
+    source = _golden() if fixture is None else fixture
+    if snapshot is None or foundation is None:
+        parsed_snapshot, parsed_foundation = _snapshot_and_foundation(source)
+        snapshot = snapshot or parsed_snapshot
+        foundation = foundation or parsed_foundation
+    selected_definition = definition or _golden_definition()
+    output_schema_bytes = canonical_json_bytes(source["output_schema"])
+    output_content_bytes = canonical_json_bytes(source["output_content"])
+    artifact = OutputArtifactRef.create(
+        schema_digest=_sha256(output_schema_bytes),
+        content_digest=_sha256(output_content_bytes),
+    )
+    return {
+        "definitions": (selected_definition,),
+        "dataset_snapshot": snapshot,
+        "foundation": foundation,
+        "selected_view_ref_ids": (VIEW_REF_ID,),
+        "input_bindings": (
+            InputBinding(
+                selected_definition.definition_id,
+                "market",
+                VIEW_REF_ID,
+                VIEW_SCHEMA_DIGEST,
+            ),
+        ),
+        "view_availability": (
+            ViewAvailability(VIEW_REF_ID, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64),
+        ),
+        "output_quality": OutputQuality(
+            "passed",
+            (OutputQualityCheck("finite_values", "passed", "sha256:" + "3" * 64),),
+        ),
+        "output_coverage": OutputCoverage(
+            "complete",
+            1,
+            1,
+            "row",
+            "alpha_005.cn_a",
+            "sha256:" + "4" * 64,
+        ),
+        "output_schema_bytes": output_schema_bytes,
+        "output_content_bytes": output_content_bytes,
+        "output_artifact_ref": artifact,
+        "availability_mode": AvailabilityMode.AS_AVAILABLE,
+        "evaluation_at": "2026-01-03T11:00:00Z",
+        "computed_at": "2026-01-03T10:15:00Z",
+        "artifact_available_at": "2026-01-03T10:20:00Z",
+        "producer": ProducerIdentity("quant_engine", "1.0.0"),
+        "code_revision": "c" * 40,
+        "actor": ActorIdentity("service", "factor_worker_v1"),
+        "correlation_id": "research_run_001",
+        "causation": Causation("foundation", foundation.foundation_id),
+        "evidence_scope": "synthetic_fixture",
+        "decision_eligible": False,
+    }
+
+
+def _factor_set(**overrides: Any) -> FactorSetRef:
+    arguments = _factor_set_arguments()
+    arguments.update(overrides)
+    return FactorSetRef.create(**arguments)
+
+
+def _assert_error(
+    error: pytest.ExceptionInfo[FactorContractError],
+    code: ContractErrorCode,
+    path: str,
+) -> None:
+    assert error.value.code is code
+    assert error.value.path == path
+
+
+def _mutate_artifact_schema_binding(value: dict[str, Any]) -> None:
+    artifact = value["output_artifact_ref"]
+    artifact["schema_digest"] = "sha256:" + "0" * 64
+    _reidentify(artifact, "artifact_id", "rhfactoroutputv1:sha256:")
+
+
+def test_golden_contracts_are_content_addressed_round_trippable_and_deeply_immutable() -> None:
+    fixture = _golden()
+    original_snapshot = copy.deepcopy(fixture["dataset_snapshot"])
+    original_foundation = copy.deepcopy(fixture["data_foundation"])
+    snapshot, foundation = _snapshot_and_foundation(fixture)
+    definition = _golden_definition()
+    factor_set = FactorSetRef.create(**_factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition))
+    binding = LegacyFactorBinding.create(
+        definition=definition,
+        legacy_factor_id="factor:demo-momentum",
+        legacy_version="1.0.0",
+        legacy_definition_sha256="b" * 64,
+        legacy_dataset_schema_version="1.0.0",
+        canonical_input_schema_digest=definition.input_schema_digest,
+        correspondence_evidence_digest="sha256:" + "5" * 64,
+    )
+
+    assert snapshot.pit_cutoff == "2026-01-02T07:01:00Z"
+    assert foundation.pit_cutoff == factor_set.pit_cutoff == "2026-01-03T00:00:00Z"
+    assert snapshot.pit_cutoff != foundation.pit_cutoff
+    assert definition.definition_id == fixture["expected"]["definition_id"]
+    assert definition.input_schema_digest == fixture["expected"]["input_schema_digest"]
+    assert factor_set.factor_set_id == fixture["expected"]["factor_set_id"]
+    assert factor_set.output_artifact_ref.artifact_id == fixture["expected"]["output_artifact_id"]
+    assert binding.binding_id == fixture["expected"]["legacy_binding_id"]
+    assert not definition.to_json().endswith("\n")
+    assert not factor_set.to_json().endswith("\n")
+    assert FactorDefinition.from_json(definition.to_json()) == definition
+
+    reparsed = FactorSetRef.from_json(
+        factor_set.to_json(),
+        definitions=(definition,),
+        dataset_snapshot=snapshot,
+        foundation=foundation,
+        output_schema_bytes=canonical_json_bytes(fixture["output_schema"]),
+        output_content_bytes=canonical_json_bytes(fixture["output_content"]),
+    )
+    reference_only = FactorSetRef.from_json(
+        factor_set.to_json(),
+        definitions=(definition,),
+        dataset_snapshot=snapshot,
+        foundation=foundation,
+    )
+    assert reparsed.factor_set_id == factor_set.factor_set_id
+    assert reparsed.payload_validation is PayloadValidation.PAYLOAD_REVALIDATED
+    assert reference_only.payload_validation is PayloadValidation.REFERENCE_ONLY
+
+    fixture["dataset_snapshot"]["descriptor"]["dataset"]["dimensions"].append("forbidden")
+    fixture["data_foundation"]["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64
+    assert snapshot.to_dict() == original_snapshot
+    assert foundation.to_dict() == original_foundation
+    returned = snapshot.to_dict()
+    returned["descriptor"]["dataset"]["dimensions"].append("also_forbidden")
+    assert snapshot.to_dict() == original_snapshot
+    with pytest.raises(FrozenInstanceError):
+        snapshot.snapshot_id = "rhdsv1:sha256:" + "0" * 64  # type: ignore[misc]
+
+
+@pytest.mark.parametrize("variant", ["whitespace", "key_order"])
+def test_contract_decoders_reject_non_canonical_json(variant: str) -> None:
+    fixture = _golden()
+    snapshot, foundation = _snapshot_and_foundation(fixture)
+    definition = _golden_definition()
+    factor_set = FactorSetRef.create(
+        **_factor_set_arguments(
+            fixture=fixture,
+            snapshot=snapshot,
+            foundation=foundation,
+            definition=definition,
+        )
+    )
+    binding = LegacyFactorBinding.create(
+        definition=definition,
+        legacy_factor_id="factor:demo-momentum",
+        legacy_version="1.0.0",
+        legacy_definition_sha256="b" * 64,
+        legacy_dataset_schema_version="1.0.0",
+        canonical_input_schema_digest=definition.input_schema_digest,
+        correspondence_evidence_digest="sha256:" + "5" * 64,
+    )
+
+    def non_canonical(value: str) -> str:
+        if variant == "whitespace":
+            return value + "\n"
+        loaded = json.loads(value)
+        reversed_items = dict(reversed(tuple(loaded.items())))
+        return json.dumps(reversed_items, ensure_ascii=False, separators=(",", ":"))
+
+    decoders = (
+        lambda value: FactorDefinition.from_json(value),
+        lambda value: FactorSetRef.from_json(
+            value,
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        ),
+        lambda value: LegacyFactorBinding.from_json(value, definition=definition),
+    )
+    for decoder, encoded in zip(
+        decoders,
+        (definition.to_json(), factor_set.to_json(), binding.to_json()),
+        strict=True,
+    ):
+        with pytest.raises(FactorContractError) as exc_info:
+            decoder(non_canonical(encoded))
+        assert exc_info.value.code is ContractErrorCode.INVALID_FORMAT
+        assert exc_info.value.path == "$"
+
+
+def test_factor_definition_identity_is_order_independent_where_semantics_are_unordered() -> None:
+    first_input = FactorInput("prices", "sha256:" + "6" * 64, ("close",))
+    second_input = FactorInput("volumes", "sha256:" + "7" * 64, ("volume",))
+    inputs = (first_input, second_input)
+    parameters_a = {
+        "window": TypedParameter("integer", 10),
+        "weights": TypedParameter("json", {"fast": [1, 2], "slow": [3, 4]}),
+    }
+    parameters_b = {
+        "weights": TypedParameter("json", {"slow": [3, 4], "fast": [1, 2]}),
+        "window": TypedParameter("integer", 10),
+    }
+    first = _definition(
+        inputs=inputs,
+        parameters=parameters_a,
+        input_schema_digest=factor_input_schema_digest(inputs),
+    )
+    second = _definition(
+        inputs=tuple(reversed(inputs)),
+        parameters=parameters_b,
+        input_schema_digest=factor_input_schema_digest(tuple(reversed(inputs))),
+    )
+    assert first.definition_id == second.definition_id
+    assert first.to_json() == second.to_json()
+
+    semantic_changes = (
+        _definition(factor_id="alpha_006"),
+        _definition(version="1.0.1"),
+        _definition(formula="correlation(close, volume, 11)"),
+        _definition(parameters={"window": TypedParameter("integer", 10)}),
+        _definition(implementation_digest="sha256:" + "9" * 64),
+        _definition(valid_until="2027-01-02T00:00:00Z"),
+        _definition(warmup_sessions=11),
+        _definition(lag_sessions=2),
+        _definition(producer=ProducerIdentity("quant_engine", "1.0.1")),
+        _definition(code_revision="d" * 40),
+    )
+    assert all(changed.definition_id != _golden_definition().definition_id for changed in semantic_changes)
+    assert len({changed.definition_id for changed in semantic_changes}) == len(semantic_changes)
+
+
+def test_parameter_types_decimal_profile_and_detached_nested_values_are_strict() -> None:
+    nested = {"ordered": [1, {"flag": True}]}
+    parameter = TypedParameter("json", nested)
+    nested["ordered"].append(2)
+    definition = _definition(parameters={"payload": parameter})
+    assert definition.to_dict()["parameters"]["payload"]["value"] == {
+        "ordered": [1, {"flag": True}]
+    }
+    integer_definition = _definition(parameters={"value": TypedParameter("integer", 1)})
+    string_definition = _definition(parameters={"value": TypedParameter("string", "1")})
+    assert integer_definition.definition_id != string_definition.definition_id
+
+    for parameter_type, value, code in (
+        ("decimal", "1.0", ContractErrorCode.INVALID_FORMAT),
+        ("decimal", "1e3", ContractErrorCode.INVALID_FORMAT),
+        ("decimal", "-0", ContractErrorCode.INVALID_FORMAT),
+        ("integer", True, ContractErrorCode.TYPE_ERROR),
+        ("json", 1.5, ContractErrorCode.TYPE_ERROR),
+        ("json", {"é": "bad-key"}, ContractErrorCode.INVALID_FORMAT),
+        ("json", 9_007_199_254_740_992, ContractErrorCode.INVALID_VALUE),
+    ):
+        with pytest.raises(FactorContractError) as error:
+            TypedParameter(parameter_type, value)
+        assert error.value.code is code
+    assert TypedParameter("decimal", "10.25").to_dict()["value"] == "10.25"
+
+
+def test_catalog_rejects_duplicate_and_overlapping_logical_validity_but_allows_adjacency() -> None:
+    base = _golden_definition()
+    adjacent = _definition(valid_from="2027-01-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z")
+    assert len(validate_factor_catalog((adjacent, base))) == 2
+    with pytest.raises(FactorContractError) as duplicate:
+        validate_factor_catalog((base, base))
+    _assert_error(duplicate, ContractErrorCode.INVALID_VALUE, "$.definitions")
+    overlapping = _definition(valid_from="2026-06-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z")
+    with pytest.raises(FactorContractError) as overlap:
+        validate_factor_catalog((base, overlapping))
+    _assert_error(overlap, ContractErrorCode.TIME_ORDER_VIOLATION, "$.definitions")
+
+
+def test_upstream_contracts_reject_unknown_fields_identity_forgery_and_unqualified_input() -> None:
+    unknown = _golden()["dataset_snapshot"]
+    unknown["provider"] = "forbidden"
+    with pytest.raises(FactorContractError) as unknown_error:
+        DatasetSnapshotEnvelope.from_dict(unknown)
+    _assert_error(unknown_error, ContractErrorCode.UNKNOWN_FIELD, "$.provider")
+
+    forged = _golden()["data_foundation"]
+    forged["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64
+    with pytest.raises(FactorContractError) as forged_error:
+        DataFoundationEnvelope.from_dict(forged)
+    assert forged_error.value.code is ContractErrorCode.IDENTITY_MISMATCH
+    assert forged_error.value.path.endswith("view_ref_id")
+
+    rejected_source = _golden()
+    rejected_source["dataset_snapshot"]["descriptor"]["qualification"]["status"] = "rejected"
+    _reidentify(rejected_source["dataset_snapshot"], "snapshot_id", "rhdsv1:sha256:")
+    rejected_snapshot = DatasetSnapshotEnvelope.from_dict(rejected_source["dataset_snapshot"])
+    _, foundation = _snapshot_and_foundation()
+    with pytest.raises(FactorContractError) as rejected_error:
+        FactorSetRef.create(
+            **_factor_set_arguments(snapshot=rejected_snapshot, foundation=foundation)
+        )
+    _assert_error(
+        rejected_error,
+        ContractErrorCode.QUALIFICATION_REJECTED,
+        "$.dataset_snapshot.descriptor.qualification",
+    )
+
+
+def test_foundation_rejects_future_knowledge_and_per_view_calendar_borrowing() -> None:
+    future = _golden()["data_foundation"]
+    action = future["corporate_action_revisions"][0]
+    old_action_id = action["action_revision_id"]
+    action["knowledge_time"] = "2026-01-03T00:00:01Z"
+    _reidentify(action, "action_revision_id", "rhcav1:sha256:")
+    future["standardized_views"][0]["corporate_action_revision_ids"] = [action["action_revision_id"]]
+    lineage = next(item for item in future["revision_lineage"] if item["revision_id"] == old_action_id)
+    lineage["revision_id"] = action["action_revision_id"]
+    lineage["knowledge_time"] = action["knowledge_time"]
+    _reidentify(future["standardized_views"][0], "view_ref_id", "rhviewrefv1:sha256:")
+    _reidentify(future, "foundation_id", "rhdfv1:sha256:")
+    with pytest.raises(FactorContractError) as future_error:
+        DataFoundationEnvelope.from_dict(future)
+    _assert_error(
+        future_error,
+        ContractErrorCode.TIME_ORDER_VIOLATION,
+        "$.revision_lineage.knowledge_time",
+    )
+
+    uncovered = _golden()["data_foundation"]
+    original_route_id = uncovered["instrument_routes"][0]["route_revision_id"]
+    second_calendar = copy.deepcopy(uncovered["trading_calendar_revisions"][0])
+    second_calendar["calendar_id"] = "rhcalendar:99990000111122223333444455556666"
+    _reidentify(second_calendar, "calendar_revision_id", "rhcalv1:sha256:")
+    uncovered["trading_calendar_revisions"].append(second_calendar)
+    route = uncovered["instrument_routes"][0]
+    route["calendar_id"] = second_calendar["calendar_id"]
+    _reidentify(route, "route_revision_id", "rhroutev1:sha256:")
+    route_lineage = next(item for item in uncovered["revision_lineage"] if item["revision_id"] == original_route_id)
+    route_lineage["revision_id"] = route["route_revision_id"]
+    uncovered["revision_lineage"].append(
+        {
+            "revision_kind": "trading_calendar",
+            "revision_id": second_calendar["calendar_revision_id"],
+            "revision_number": 1,
+            "knowledge_time": second_calendar["knowledge_time"],
+            "evidence_digest": second_calendar["evidence_digest"],
+        }
+    )
+    view = uncovered["standardized_views"][0]
+    view["instrument_route_revision_ids"] = [route["route_revision_id"]]
+    _reidentify(view, "view_ref_id", "rhviewrefv1:sha256:")
+    _reidentify(uncovered, "foundation_id", "rhdfv1:sha256:")
+    with pytest.raises(FactorContractError) as calendar_error:
+        DataFoundationEnvelope.from_dict(uncovered)
+    assert calendar_error.value.code is ContractErrorCode.INPUT_CLOSURE_VIOLATION
+    assert "selected route calendar" in calendar_error.value.detail
+
+
+def _replay_fixture() -> dict[str, Any]:
+    fixture = _golden()
+    snapshot = fixture["dataset_snapshot"]
+    snapshot["descriptor"]["published_at"] = "2026-01-04T00:00:00Z"
+    _reidentify(snapshot, "snapshot_id", "rhdsv1:sha256:")
+    foundation = fixture["data_foundation"]
+    foundation["dataset_snapshot_id"] = snapshot["snapshot_id"]
+    for view in foundation["standardized_views"]:
+        view["dataset_snapshot_id"] = snapshot["snapshot_id"]
+        _reidentify(view, "view_ref_id", "rhviewrefv1:sha256:")
+    _reidentify(foundation, "foundation_id", "rhdfv1:sha256:")
+    return fixture
+
+
+def test_as_available_and_retrospective_replay_keep_distinct_time_claims() -> None:
+    as_available = _factor_set()
+    assert as_available.historical_availability is HistoricalAvailability.DECLARED_AS_AVAILABLE
+
+    replay_source = _replay_fixture()
+    snapshot, foundation = _snapshot_and_foundation(replay_source)
+    replay_view_id = next(iter(foundation.views))
+    arguments = _factor_set_arguments(
+        fixture=replay_source,
+        snapshot=snapshot,
+        foundation=foundation,
+    )
+    arguments.update(
+        selected_view_ref_ids=(replay_view_id,),
+        input_bindings=(
+            InputBinding(
+                arguments["definitions"][0].definition_id,
+                "market",
+                replay_view_id,
+                VIEW_SCHEMA_DIGEST,
+            ),
+        ),
+        view_availability=(
+            ViewAvailability(replay_view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),
+        ),
+        availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY,
+        computed_at="2026-01-04T00:20:00Z",
+        artifact_available_at="2026-01-04T00:25:00Z",
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    replay = FactorSetRef.create(**arguments)
+    assert replay.evaluation_at == "2026-01-03T11:00:00Z"
+    assert replay.computed_at == "2026-01-04T00:20:00Z"
+    assert replay.historical_availability is HistoricalAvailability.NOT_ESTABLISHED
+
+    replay_source_args = _factor_set_arguments(
+        fixture=replay_source,
+        snapshot=snapshot,
+        foundation=foundation,
+    )
+    replay_source_args.update(
+        selected_view_ref_ids=(replay_view_id,),
+        input_bindings=(
+            InputBinding(
+                replay_source_args["definitions"][0].definition_id,
+                "market",
+                replay_view_id,
+                VIEW_SCHEMA_DIGEST,
+            ),
+        ),
+        view_availability=(
+            ViewAvailability(replay_view_id, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64),
+        ),
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    with pytest.raises(FactorContractError) as late_publication:
+        FactorSetRef.create(**replay_source_args)
+    _assert_error(
+        late_publication,
+        ContractErrorCode.TIME_ORDER_VIOLATION,
+        "$.dataset_snapshot.descriptor.published_at",
+    )
+
+
+@pytest.mark.parametrize(
+    ("overrides", "path"),
+    [
+        ({"view_availability": (ViewAvailability(VIEW_REF_ID, "2026-01-03T00:00:01Z", "sha256:" + "2" * 64),)}, "$.view_availability"),
+        ({"computed_at": "2026-01-02T23:40:00Z"}, "$.computed_at"),
+        ({"artifact_available_at": "2026-01-03T10:14:00Z"}, "$.artifact_available_at"),
+        ({"artifact_available_at": "2026-01-03T11:00:01Z"}, "$.artifact_available_at"),
+        ({"evaluation_at": "2026-01-03T11:00:00"}, "$.evaluation_at"),
+    ],
+)
+def test_as_available_time_failures_are_typed(overrides: dict[str, Any], path: str) -> None:
+    with pytest.raises(FactorContractError) as error:
+        _factor_set(**overrides)
+    assert error.value.code in {
+        ContractErrorCode.INVALID_FORMAT,
+        ContractErrorCode.TIME_ORDER_VIOLATION,
+    }
+    assert error.value.path == path
+
+
+def test_replay_rejects_backdating_and_historical_availability_promotion() -> None:
+    source = _replay_fixture()
+    snapshot, foundation = _snapshot_and_foundation(source)
+    view_id = next(iter(foundation.views))
+    arguments = _factor_set_arguments(fixture=source, snapshot=snapshot, foundation=foundation)
+    definition = arguments["definitions"][0]
+    arguments.update(
+        selected_view_ref_ids=(view_id,),
+        input_bindings=(InputBinding(definition.definition_id, "market", view_id, VIEW_SCHEMA_DIGEST),),
+        view_availability=(ViewAvailability(view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),),
+        availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY,
+        computed_at="2026-01-04T00:20:00Z",
+        artifact_available_at="2026-01-04T00:25:00Z",
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    replay = FactorSetRef.create(**arguments)
+    promoted = replay.to_dict()
+    promoted["historical_availability"] = "declared_as_available"
+    _reidentify(promoted, "factor_set_id", "rhfactorsetv1:sha256:")
+    with pytest.raises(FactorContractError) as promotion_error:
+        FactorSetRef.from_dict(
+            promoted,
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(
+        promotion_error,
+        ContractErrorCode.READINESS_ESCALATION,
+        "$.historical_availability",
+    )
+    arguments["computed_at"] = "2026-01-03T11:30:00Z"
+    with pytest.raises(FactorContractError) as backdated_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(backdated_error, ContractErrorCode.TIME_ORDER_VIOLATION, "$.computed_at")
+
+
+def _multi_view_fixture() -> tuple[dict[str, Any], str]:
+    fixture = _golden()
+    foundation = fixture["data_foundation"]
+    second = copy.deepcopy(foundation["standardized_views"][0])
+    second["view_id"] = "rhview:11111111222222223333333344444444"
+    second["schema_digest"] = "sha256:" + "6" * 64
+    second["content_digest"] = "sha256:" + "7" * 64
+    second["transformation_digest"] = "sha256:" + "8" * 64
+    _reidentify(second, "view_ref_id", "rhviewrefv1:sha256:")
+    foundation["standardized_views"].append(second)
+    _reidentify(foundation, "foundation_id", "rhdfv1:sha256:")
+    return fixture, second["view_ref_id"]
+
+
+def test_multi_input_mapping_requires_exact_consumption_closure_and_is_order_independent() -> None:
+    fixture, second_view_id = _multi_view_fixture()
+    snapshot, foundation = _snapshot_and_foundation(fixture)
+    inputs = (
+        FactorInput("prices", VIEW_SCHEMA_DIGEST, ("close",)),
+        FactorInput("volumes", "sha256:" + "6" * 64, ("volume",)),
+    )
+    definition = _definition(
+        inputs=inputs,
+        formula="correlation(close, volume, 10)",
+        input_schema_digest=factor_input_schema_digest(inputs),
+    )
+    first_binding = InputBinding(definition.definition_id, "prices", VIEW_REF_ID, VIEW_SCHEMA_DIGEST)
+    second_binding = InputBinding(definition.definition_id, "volumes", second_view_id, "sha256:" + "6" * 64)
+    first_availability = ViewAvailability(VIEW_REF_ID, "2026-01-02T23:40:00Z", "sha256:" + "2" * 64)
+    second_availability = ViewAvailability(second_view_id, "2026-01-02T23:50:00Z", "sha256:" + "6" * 64)
+    base = _factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition)
+    base.update(
+        selected_view_ref_ids=(VIEW_REF_ID, second_view_id),
+        input_bindings=(first_binding, second_binding),
+        view_availability=(first_availability, second_availability),
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    first = FactorSetRef.create(**base)
+    reordered = dict(base)
+    reordered.update(
+        selected_view_ref_ids=(second_view_id, VIEW_REF_ID),
+        input_bindings=(second_binding, first_binding),
+        view_availability=(second_availability, first_availability),
+    )
+    assert FactorSetRef.create(**reordered).factor_set_id == first.factor_set_id
+
+    for invalid_bindings, invalid_views in (
+        ((first_binding,), (VIEW_REF_ID, second_view_id)),
+        ((first_binding, second_binding), (VIEW_REF_ID,)),
+        ((first_binding, second_binding), (VIEW_REF_ID, second_view_id, VIEW_REF_ID)),
+    ):
+        invalid = dict(base)
+        invalid.update(input_bindings=invalid_bindings, selected_view_ref_ids=invalid_views)
+        with pytest.raises(FactorContractError) as error:
+            FactorSetRef.create(**invalid)
+        assert error.value.code in {
+            ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+            ContractErrorCode.INVALID_VALUE,
+        }
+
+
+@pytest.mark.parametrize(
+    ("mutate", "code", "path"),
+    [
+        (lambda value: value["producer"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.producer.id"),
+        (lambda value: value["producer"].pop("version"), ContractErrorCode.MISSING_FIELD, "$.producer.version"),
+        (lambda value: value["producer"].update(id="other_engine"), ContractErrorCode.LINEAGE_VIOLATION, "$.producer.id"),
+        (lambda value: value["producer"].update(version="latest"), ContractErrorCode.INVALID_FORMAT, "$.producer.version"),
+        (lambda value: value.update(code_revision="bad"), ContractErrorCode.INVALID_FORMAT, "$.code_revision"),
+        (lambda value: value["actor"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.actor.kind"),
+        (lambda value: value["actor"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.actor.id"),
+        (lambda value: value["actor"].update(kind="robot"), ContractErrorCode.INVALID_VALUE, "$.actor.kind"),
+        (lambda value: value["actor"].update(id="latest"), ContractErrorCode.INVALID_VALUE, "$.actor.id"),
+        (lambda value: value.pop("correlation_id"), ContractErrorCode.MISSING_FIELD, "$.correlation_id"),
+        (lambda value: value.update(correlation_id="latest"), ContractErrorCode.INVALID_VALUE, "$.correlation_id"),
+        (lambda value: value["causation"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.causation.kind"),
+        (lambda value: value["causation"].update(kind="run"), ContractErrorCode.INVALID_VALUE, "$.causation.kind"),
+        (lambda value: value["causation"].update(id="rhdfv1:sha256:" + "0" * 64), ContractErrorCode.LINEAGE_VIOLATION, "$.causation.id"),
+        (lambda value: value.pop("output_artifact_ref"), ContractErrorCode.MISSING_FIELD, "$.output_artifact_ref"),
+        (lambda value: value["output_artifact_ref"].update(artifact_id="rhfactoroutputv1:sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.output_artifact_ref.artifact_id"),
+        (_mutate_artifact_schema_binding, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref"),
+        (lambda value: value.update(availability_mode="implicit_fallback"), ContractErrorCode.INVALID_VALUE, "$.availability_mode"),
+        (lambda value: value.pop("computed_at"), ContractErrorCode.MISSING_FIELD, "$.computed_at"),
+        (lambda value: value.update(decision_eligible=True), ContractErrorCode.READINESS_ESCALATION, "$.decision_eligible"),
+        (lambda value: value.update(evidence_scope="real_data"), ContractErrorCode.READINESS_ESCALATION, "$.evidence_scope"),
+        (lambda value: value["upstream_evidence"].update(qualification_evidence_digest="sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.upstream_evidence"),
+    ],
+)
+def test_lineage_artifact_and_readiness_fields_have_independent_typed_negatives(
+    mutate: Callable[[dict[str, Any]], Any],
+    code: ContractErrorCode,
+    path: str,
+) -> None:
+    factor_set = _factor_set()
+    value = factor_set.to_dict()
+    mutate(value)
+    if "factor_set_id" in value:
+        _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    snapshot, foundation = _snapshot_and_foundation()
+    with pytest.raises(FactorContractError) as error:
+        FactorSetRef.from_dict(
+            value,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(error, code, path)
+
+
+def test_output_schema_content_bytes_cannot_be_swapped_or_forged() -> None:
+    factor_set = _factor_set()
+    fixture = _golden()
+    snapshot, foundation = _snapshot_and_foundation()
+    schema_bytes = canonical_json_bytes(fixture["output_schema"])
+    content_bytes = canonical_json_bytes(fixture["output_content"])
+    with pytest.raises(FactorContractError) as swapped:
+        FactorSetRef.from_dict(
+            factor_set.to_dict(),
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+            output_schema_bytes=content_bytes,
+            output_content_bytes=schema_bytes,
+        )
+    _assert_error(swapped, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref")
+    with pytest.raises(FactorContractError) as noncanonical:
+        FactorSetRef.create(
+            **{
+                **_factor_set_arguments(),
+                "output_schema_bytes": json.dumps(fixture["output_schema"], indent=2).encode(),
+            }
+        )
+    _assert_error(noncanonical, ContractErrorCode.INVALID_FORMAT, "$.output_schema_bytes")
+
+
+def test_unsuccessful_output_quality_or_coverage_cannot_form_a_factor_set() -> None:
+    with pytest.raises(FactorContractError) as failed_quality:
+        _factor_set(
+            output_quality=OutputQuality(
+                "failed",
+                (OutputQualityCheck("finite_values", "failed", "sha256:" + "3" * 64),),
+            )
+        )
+    _assert_error(failed_quality, ContractErrorCode.INVALID_VALUE, "$.output_quality")
+
+    for coverage in (
+        OutputCoverage("incomplete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64),
+        OutputCoverage("complete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64),
+    ):
+        with pytest.raises(FactorContractError) as incomplete:
+            _factor_set(output_coverage=coverage)
+        _assert_error(incomplete, ContractErrorCode.INVALID_VALUE, "$.output_coverage")
+
+
+def test_external_snapshot_definition_and_view_references_cannot_be_substituted() -> None:
+    factor_set = _factor_set()
+    snapshot, foundation = _snapshot_and_foundation()
+    value = factor_set.to_dict()
+    value["dataset_snapshot_id"] = "rhdsv1:sha256:" + "0" * 64
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    with pytest.raises(FactorContractError) as snapshot_error:
+        FactorSetRef.from_dict(
+            value,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(
+        snapshot_error,
+        ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.dataset_snapshot_id",
+    )
+
+    value = factor_set.to_dict()
+    value["definition_ids"] = ["rhfactorv1:sha256:" + "0" * 64]
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    with pytest.raises(FactorContractError) as definition_error:
+        FactorSetRef.from_dict(
+            value,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(
+        definition_error,
+        ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.definition_ids",
+    )
+
+    arguments = _factor_set_arguments()
+    arguments["selected_view_ref_ids"] = ("rhviewrefv1:sha256:" + "0" * 64,)
+    with pytest.raises(FactorContractError) as view_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(
+        view_error,
+        ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.selected_view_ref_ids",
+    )
+
+
+@pytest.mark.parametrize(
+    "invalid_definition_id",
+    [
+        {"unexpected": "object"},
+        ["array"],
+        42,
+        True,
+        None,
+    ],
+)
+def test_factor_set_ref_definition_ids_reject_non_string_types(
+    invalid_definition_id: Any,
+) -> None:
+    factor_set = _factor_set()
+    definition = _golden_definition()
+    value = factor_set.to_dict()
+    value["definition_ids"] = [definition.definition_id, invalid_definition_id]
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    snapshot, foundation = _snapshot_and_foundation()
+
+    with pytest.raises(FactorContractError) as error:
+        FactorSetRef.from_json(
+            canonical_json_bytes(value),
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+
+    _assert_error(error, ContractErrorCode.TYPE_ERROR, "$.definition_ids[1]")
+
+
+def test_factor_set_ref_definition_ids_still_reject_duplicate_strings() -> None:
+    factor_set = _factor_set()
+    definition = _golden_definition()
+    value = factor_set.to_dict()
+    value["definition_ids"] = [definition.definition_id, definition.definition_id]
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    snapshot, foundation = _snapshot_and_foundation()
+
+    with pytest.raises(FactorContractError) as error:
+        FactorSetRef.from_json(
+            canonical_json_bytes(value),
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+
+    _assert_error(error, ContractErrorCode.INVALID_VALUE, "$.definition_ids")
+
+
+def test_factor_set_parent_requires_exact_identity_and_correlation() -> None:
+    parent = _factor_set()
+    child_arguments = _factor_set_arguments()
+    child_arguments.update(
+        output_content_bytes=canonical_json_bytes({"rows": [{"value": "0.250"}]}),
+        causation=Causation("factor_set", parent.factor_set_id),
+        parent=parent,
+    )
+    child_arguments["output_artifact_ref"] = OutputArtifactRef.create(
+        schema_digest=_sha256(child_arguments["output_schema_bytes"]),
+        content_digest=_sha256(child_arguments["output_content_bytes"]),
+    )
+    child = FactorSetRef.create(**child_arguments)
+    assert child.causation.id == parent.factor_set_id
+    missing_parent = child.to_dict()
+    snapshot, foundation = _snapshot_and_foundation()
+    with pytest.raises(FactorContractError) as missing_error:
+        FactorSetRef.from_dict(
+            missing_parent,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(missing_error, ContractErrorCode.LINEAGE_VIOLATION, "$.causation")
+    wrong_correlation = dict(child_arguments)
+    wrong_correlation["correlation_id"] = "different_run"
+    with pytest.raises(FactorContractError) as correlation_error:
+        FactorSetRef.create(**wrong_correlation)
+    _assert_error(correlation_error, ContractErrorCode.LINEAGE_VIOLATION, "$.correlation_id")
+
+
+def test_legacy_bridge_is_explicit_lossy_and_preserves_all_four_historical_fields() -> None:
+    definition = _golden_definition()
+    legacy = FactorVersion(
+        factor_id="factor:demo-momentum",
+        version="1.0.0",
+        definition_sha256="b" * 64,
+        dataset_schema_version="1.0.0",
+    )
+    binding = LegacyFactorBinding.create(
+        definition=definition,
+        legacy_factor_id=legacy.factor_id,
+        legacy_version=legacy.version,
+        legacy_definition_sha256=legacy.definition_sha256,
+        legacy_dataset_schema_version=legacy.dataset_schema_version,
+        canonical_input_schema_digest=definition.input_schema_digest,
+        correspondence_evidence_digest="sha256:" + "5" * 64,
+    )
+    assert bind_legacy_factor(legacy, definition, binding) is definition
+    assert project_legacy_factor(definition, binding) == legacy
+    assert legacy.version_id == "factor:demo-momentum@1.0.0"
+    assert legacy.definition_sha256 != definition.definition_id.rsplit(":", maxsplit=1)[-1]
+    assert LegacyFactorBinding.from_json(binding.to_json(), definition=definition) == binding
+
+    mismatched = FactorVersion(
+        factor_id="factor:different",
+        version=legacy.version,
+        definition_sha256=legacy.definition_sha256,
+        dataset_schema_version=legacy.dataset_schema_version,
+    )
+    with pytest.raises(FactorContractError) as mismatch_error:
+        bind_legacy_factor(mismatched, definition, binding)
+    _assert_error(mismatch_error, ContractErrorCode.LEGACY_BINDING_MISMATCH, "$.binding")
+
+
+def test_bare_legacy_factor_or_id_cannot_enter_factor_set_contract() -> None:
+    legacy = FactorVersion("factor:demo-momentum", "1.0.0", "b" * 64, "1.0.0")
+    arguments = _factor_set_arguments()
+    arguments["definitions"] = (legacy,)
+    with pytest.raises(FactorContractError) as legacy_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(legacy_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")
+    arguments["definitions"] = (legacy.version_id,)
+    with pytest.raises(FactorContractError) as id_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(id_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")

tests/test_governed_pipeline.py

@@ -0,0 +1,338 @@
+"""Governed Personal Quant OS vertical-slice contracts."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pandas as pd
+import pytest
+
+from quant_engine.execution import ExecutionConfig
+from quant_engine.governed_pipeline import (
+    DatasetSnapshot,
+    FactorVersion,
+    PaperOrderIntent,
+    RiskDecisionStatus,
+    RiskPolicy,
+    StrategyStage,
+    StrategyVersion,
+    create_paper_order_intent,
+    run_governed_factor_slice,
+)
+
+
+def _calendar() -> pd.DatetimeIndex:
+    return pd.date_range("2026-01-05", periods=4, freq="B")
+
+
+def _scores() -> pd.DataFrame:
+    dates = _calendar()
+    return pd.DataFrame(
+        {"A": [3.0, 1.0], "B": [2.0, 3.0], "C": [1.0, 2.0]},
+        index=dates[:2],
+    )
+
+
+def _prices() -> tuple[pd.DataFrame, pd.DataFrame]:
+    dates = _calendar()
+    opens = pd.DataFrame(
+        {"A": [10.0, 10.0, 10.2, 10.4], "B": [20.0, 20.0, 20.5, 21.0], "C": [30.0, 30.0, 30.0, 30.0]},
+        index=dates,
+    )
+    closes = opens * 1.01
+    return opens, closes
+
+
+def _snapshot() -> DatasetSnapshot:
+    return DatasetSnapshot(
+        snapshot_id="dataset:cn-a-daily-20260108-v1",
+        schema_version="1.0.0",
+        content_sha256="a" * 64,
+        effective_at=datetime(2026, 1, 8, 7, tzinfo=UTC),
+        available_at=datetime(2026, 1, 8, 8, tzinfo=UTC),
+        ingested_at=datetime(2026, 1, 8, 8, 5, tzinfo=UTC),
+    )
+
+
+def _factor() -> FactorVersion:
+    return FactorVersion(
+        factor_id="factor:demo-momentum",
+        version="1.0.0",
+        definition_sha256="b" * 64,
+        dataset_schema_version="1.0.0",
+    )
+
+
+def _strategy() -> StrategyVersion:
+    return StrategyVersion(
+        strategy_id="strategy:demo-top2",
+        version="1.0.0",
+        factor_version_id="factor:demo-momentum@1.0.0",
+        stage=StrategyStage.APPROVED,
+    )
+
+
+def _execution_config() -> ExecutionConfig:
+    return ExecutionConfig(
+        commission_bps=0,
+        stamp_tax_bps=0,
+        slippage_bps=0,
+        min_trade_amount=0,
+    )
+
+
+def test_governed_slice_is_reproducible_and_creates_only_paper_intent() -> None:
+    opens, closes = _prices()
+    created_at = datetime(2026, 1, 9, 1, tzinfo=UTC)
+    policy = RiskPolicy(
+        policy_id="risk:paper-default@1.0.0",
+        max_gross_exposure=1.0,
+        max_single_asset_weight=0.6,
+        max_positions=10,
+    )
+
+    result = run_governed_factor_slice(
+        factor_scores=_scores(),
+        execution_prices=opens,
+        valuation_prices=closes,
+        dataset_snapshot=_snapshot(),
+        factor_version=_factor(),
+        strategy_version=_strategy(),
+        risk_policy=policy,
+        code_revision="c" * 40,
+        created_at=created_at,
+        top_k=2,
+        execution_price_field="open",
+        valuation_price_field="close",
+        execution_config=_execution_config(),
+    )
+
+    assert result.backtest_run.dataset_snapshot_id == _snapshot().snapshot_id
+    assert result.backtest_run.factor_version_id == _factor().version_id
+    assert result.backtest_run.strategy_version_id == _strategy().version_id
+    assert result.backtest_run.code_revision == "c" * 40
+    assert len(result.backtest_run.config_hash) == 64
+    assert result.portfolio_target.backtest_run_id == result.backtest_run.run_id
+    assert result.risk_decision.status is RiskDecisionStatus.APPROVED
+    assert result.risk_decision.portfolio_target_id == result.portfolio_target.target_id
+    assert result.order_intent is not None
+    assert result.order_intent.environment == "paper"
+    assert result.order_intent.risk_decision_id == result.risk_decision.decision_id
+    assert result.order_intent.portfolio_target_id == result.portfolio_target.target_id
+    assert result.factor_version.version_id == "factor:demo-momentum@1.0.0"
+    assert result.factor_version.definition_sha256 == "b" * 64
+    assert result.strategy_version.version_id == "strategy:demo-top2@1.0.0"
+    assert result.backtest_run.run_id == (
+        "backtest-run:e74403571f6a73c98b380220748957a422518c0bed883fabc4b93ebe13f05a37"
+    )
+    assert result.backtest_run.config_hash == (
+        "40a3c804a2dc940161a626d1a5d25817c13463e685005e37fc48c41d1e20b87b"
+    )
+    assert result.portfolio_target.target_id == (
+        "portfolio-target:ab2d398489aa9a292ee155a1098e9340beac0a924874cdaeb5b7b4d379ac9ce8"
+    )
+    assert result.risk_decision.decision_id == (
+        "risk-decision:95924926bd327e44beeb15a63f47d14c5e77b97533fc3227fba2eda68e9b423d"
+    )
+    assert result.order_intent.intent_id == (
+        "order-intent:73c349086c2c05b68424ace9286896eb21507b9080da5ff9b96b58c88d8f6ac1"
+    )
+
+    repeated = run_governed_factor_slice(
+        factor_scores=_scores(),
+        execution_prices=opens,
+        valuation_prices=closes,
+        dataset_snapshot=_snapshot(),
+        factor_version=_factor(),
+        strategy_version=_strategy(),
+        risk_policy=policy,
+        code_revision="c" * 40,
+        created_at=created_at,
+        top_k=2,
+        execution_price_field="open",
+        valuation_price_field="close",
+        execution_config=_execution_config(),
+    )
+    assert repeated.backtest_run.run_id == result.backtest_run.run_id
+    assert repeated.portfolio_target.target_id == result.portfolio_target.target_id
+    assert repeated.risk_decision.decision_id == result.risk_decision.decision_id
+    assert repeated.order_intent == result.order_intent
+
+
+def test_risk_rejection_blocks_order_intent() -> None:
+    opens, closes = _prices()
+    result = run_governed_factor_slice(
+        factor_scores=_scores(),
+        execution_prices=opens,
+        valuation_prices=closes,
+        dataset_snapshot=_snapshot(),
+        factor_version=_factor(),
+        strategy_version=_strategy(),
+        risk_policy=RiskPolicy(
+            policy_id="risk:no-concentration@1.0.0",
+            max_gross_exposure=1.0,
+            max_single_asset_weight=0.4,
+            max_positions=10,
+        ),
+        code_revision="c" * 40,
+        created_at=datetime(2026, 1, 9, 1, tzinfo=UTC),
+        top_k=2,
+        execution_price_field="open",
+        valuation_price_field="close",
+        execution_config=_execution_config(),
+    )
+
+    assert result.risk_decision.status is RiskDecisionStatus.REJECTED
+    assert any("single-asset weight" in reason for reason in result.risk_decision.reasons)
+    assert result.order_intent is None
+    with pytest.raises(ValueError, match="approved risk decision"):
+        create_paper_order_intent(result.portfolio_target, result.risk_decision)
+    with pytest.raises(ValueError, match="approved risk decision"):
+        PaperOrderIntent(result.portfolio_target, result.risk_decision)
+
+
+def test_dataset_snapshot_requires_point_in_time_ordering_and_aware_times() -> None:
+    with pytest.raises(ValueError, match="timezone-aware"):
+        DatasetSnapshot(
+            snapshot_id="dataset:invalid",
+            schema_version="1.0.0",
+            content_sha256="a" * 64,
+            effective_at=datetime(2026, 1, 8, 7),
+            available_at=datetime(2026, 1, 8, 8, tzinfo=UTC),
+            ingested_at=datetime(2026, 1, 8, 9, tzinfo=UTC),
+        )
+
+    with pytest.raises(ValueError, match="effective_at <= available_at <= ingested_at"):
+        DatasetSnapshot(
+            snapshot_id="dataset:invalid",
+            schema_version="1.0.0",
+            content_sha256="a" * 64,
+            effective_at=datetime(2026, 1, 8, 9, tzinfo=UTC),
+            available_at=datetime(2026, 1, 8, 8, tzinfo=UTC),
+            ingested_at=datetime(2026, 1, 8, 10, tzinfo=UTC),
+        )
+
+
+def test_strategy_factor_lineage_must_match() -> None:
+    opens, closes = _prices()
+    mismatched = StrategyVersion(
+        strategy_id="strategy:demo-top2",
+        version="1.0.0",
+        factor_version_id="factor:other@1.0.0",
+        stage=StrategyStage.APPROVED,
+    )
+
+    with pytest.raises(ValueError, match="factor lineage"):
+        run_governed_factor_slice(
+            factor_scores=_scores(),
+            execution_prices=opens,
+            valuation_prices=closes,
+            dataset_snapshot=_snapshot(),
+            factor_version=_factor(),
+            strategy_version=mismatched,
+            risk_policy=RiskPolicy(
+                policy_id="risk:paper-default@1.0.0",
+                max_gross_exposure=1.0,
+                max_single_asset_weight=0.6,
+                max_positions=10,
+            ),
+            code_revision="c" * 40,
+            created_at=datetime(2026, 1, 9, 1, tzinfo=UTC),
+            top_k=2,
+            execution_price_field="open",
+            valuation_price_field="close",
+            execution_config=_execution_config(),
+        )
+
+
+def test_governed_slice_requires_matching_schema_and_snapshot_available_by_run_time() -> None:
+    opens, closes = _prices()
+    common = {
+        "factor_scores": _scores(),
+        "execution_prices": opens,
+        "valuation_prices": closes,
+        "strategy_version": _strategy(),
+        "risk_policy": RiskPolicy(
+            policy_id="risk:paper-default@1.0.0",
+            max_gross_exposure=1.0,
+            max_single_asset_weight=0.6,
+            max_positions=10,
+        ),
+        "code_revision": "c" * 40,
+        "top_k": 2,
+        "execution_price_field": "open",
+        "valuation_price_field": "close",
+        "execution_config": _execution_config(),
+    }
+
+    with pytest.raises(ValueError, match="dataset schema"):
+        run_governed_factor_slice(
+            dataset_snapshot=_snapshot(),
+            factor_version=FactorVersion(
+                factor_id="factor:demo-momentum",
+                version="1.0.0",
+                definition_sha256="b" * 64,
+                dataset_schema_version="2.0.0",
+            ),
+            created_at=datetime(2026, 1, 9, 1, tzinfo=UTC),
+            **common,
+        )
+
+    with pytest.raises(ValueError, match="available before the research run"):
+        run_governed_factor_slice(
+            dataset_snapshot=_snapshot(),
+            factor_version=_factor(),
+            created_at=datetime(2026, 1, 8, 7, 30, tzinfo=UTC),
+            **common,
+        )
+
+    future_scores = _scores()
+    future_scores.index = pd.date_range("2026-01-12", periods=2, freq="B")
+    with pytest.raises(ValueError, match="future decision dates"):
+        run_governed_factor_slice(
+            dataset_snapshot=_snapshot(),
+            factor_version=_factor(),
+            factor_scores=future_scores,
+            execution_prices=opens,
+            valuation_prices=closes,
+            strategy_version=common["strategy_version"],
+            risk_policy=common["risk_policy"],
+            code_revision="c" * 40,
+            created_at=datetime(2026, 1, 9, 1, tzinfo=UTC),
+            top_k=2,
+            execution_price_field="open",
+            valuation_price_field="close",
+            execution_config=_execution_config(),
+        )
+
+
+def test_paper_intent_requires_approved_strategy_stage() -> None:
+    opens, closes = _prices()
+    validated = StrategyVersion(
+        strategy_id="strategy:demo-top2",
+        version="1.0.0",
+        factor_version_id=_factor().version_id,
+        stage=StrategyStage.VALIDATED,
+    )
+
+    with pytest.raises(ValueError, match="Approved or Paper"):
+        run_governed_factor_slice(
+            factor_scores=_scores(),
+            execution_prices=opens,
+            valuation_prices=closes,
+            dataset_snapshot=_snapshot(),
+            factor_version=_factor(),
+            strategy_version=validated,
+            risk_policy=RiskPolicy(
+                policy_id="risk:paper-default@1.0.0",
+                max_gross_exposure=1.0,
+                max_single_asset_weight=0.6,
+                max_positions=10,
+            ),
+            code_revision="c" * 40,
+            created_at=datetime(2026, 1, 9, 1, tzinfo=UTC),
+            top_k=2,
+            execution_price_field="open",
+            valuation_price_field="close",
+            execution_config=_execution_config(),
+        )