quant_engine/tests/test_governed_pipeline.py

"""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(),
        )