diff --git a/README.md b/README.md index 3ac93af..42f488c 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ - `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 +56,9 @@ pytest # 单元测试 pytest --cov=src # 覆盖率 mypy --strict src/ # 类型检查 ruff check src/ tests/ # lint + +# 无网络、无数据库、无券商的架构烟测 +uv run python -m quant_engine.governed_pipeline ``` ## 使用 @@ -191,6 +195,23 @@ print(backtest.stats()) print(backtest.benchmark_report()) ``` +## 治理垂直切片 + +`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 保持向后兼容）： diff --git a/src/quant_engine/governed_pipeline.py b/src/quant_engine/governed_pipeline.py new file mode 100644 index 0000000..28d1080 --- /dev/null +++ b/src/quant_engine/governed_pipeline.py @@ -0,0 +1,540 @@ +"""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.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", + "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}" + + +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)) diff --git a/tests/test_governed_pipeline.py b/tests/test_governed_pipeline.py new file mode 100644 index 0000000..466cf8b --- /dev/null +++ b/tests/test_governed_pipeline.py @@ -0,0 +1,320 @@ +"""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 + + 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(), + )