quant-os/tests/test_baseline_pipeline.py

import copy
import unittest

from quant60.baseline_pipeline import (
    AlphaForecastInput,
    BaselinePipelineError,
    BaselinePipelinePolicy,
    CostForecastInput,
    FIVE_LAYER_ORDER,
    RiskForecastInput,
    SecurityStateInput,
    build_baseline_pipeline,
    verify_layer_trace,
)
from quant60.optimizer import PortfolioConstraints
from quant60.universe import UniverseState


SYMBOLS = (
    "600000.XSHG",
    "000001.XSHE",
    "000333.XSHE",
)


def inputs(*, first_variance=0.03, first_cost_bps=8.0):
    alpha = [
        AlphaForecastInput(
            SYMBOLS[0],
            expected_excess_return=0.030,
            model_version="ridge-frozen-fixture-v1",
            feature_version="transparent-price-v1",
        ),
        AlphaForecastInput(
            SYMBOLS[1],
            expected_excess_return=0.020,
            model_version="ridge-frozen-fixture-v1",
            feature_version="transparent-price-v1",
        ),
        AlphaForecastInput(
            SYMBOLS[2],
            expected_excess_return=0.010,
            model_version="ridge-frozen-fixture-v1",
            feature_version="transparent-price-v1",
        ),
    ]
    risk = [
        RiskForecastInput(
            SYMBOLS[0],
            first_variance,
            "diagonal-risk-fixture-v1",
        ),
        RiskForecastInput(SYMBOLS[1], 0.03, "diagonal-risk-fixture-v1"),
        RiskForecastInput(SYMBOLS[2], 0.03, "diagonal-risk-fixture-v1"),
    ]
    cost = [
        CostForecastInput(
            SYMBOLS[0],
            first_cost_bps,
            1.0,
            "dated-cost-fixture-v1",
        ),
        CostForecastInput(SYMBOLS[1], 8.0, 1.0, "dated-cost-fixture-v1"),
        CostForecastInput(SYMBOLS[2], 8.0, 1.0, "dated-cost-fixture-v1"),
    ]
    securities = [
        SecurityStateInput(
            SYMBOLS[0],
            UniverseState.OPENABLE,
            "bank",
            10.0,
        ),
        SecurityStateInput(
            SYMBOLS[1],
            UniverseState.OPENABLE,
            "bank",
            20.0,
        ),
        SecurityStateInput(
            SYMBOLS[2],
            UniverseState.OPENABLE,
            "consumer",
            50.0,
        ),
    ]
    return alpha, risk, cost, securities


def policy():
    return BaselinePipelinePolicy(
        constraints=PortfolioConstraints(
            gross_target=0.80,
            single_name_cap=0.80,
            industry_cap=0.80,
            one_way_turnover_cap=0.80,
            risk_aversion=1.0,
            cost_aversion=1.0,
            turnover_aversion=0.0,
        ),
        cash_buffer=0.10,
    )


class BaselinePipelineTests(unittest.TestCase):
    def test_canonical_path_emits_exact_five_layer_hash_chain(self):
        alpha, risk, cost, securities = inputs()
        result = build_baseline_pipeline(
            alpha=alpha,
            risk=risk,
            cost=cost,
            securities=securities,
            equity=1_000_000,
            policy=policy(),
        )
        self.assertEqual(
            tuple(event["layer"] for event in result.layer_trace),
            FIVE_LAYER_ORDER,
        )
        self.assertTrue(result.risk_decision.approved)
        self.assertTrue(result.order_deltas)
        verify_layer_trace(result.layer_trace, result.layer_payloads)

    def test_risk_and_cost_are_causal_inputs_to_portfolio(self):
        alpha, risk, cost, securities = inputs()
        base = build_baseline_pipeline(
            alpha=alpha,
            risk=risk,
            cost=cost,
            securities=securities,
            equity=1_000_000,
            policy=policy(),
        )
        alpha, risk, cost, securities = inputs(first_variance=0.30)
        high_risk = build_baseline_pipeline(
            alpha=alpha,
            risk=risk,
            cost=cost,
            securities=securities,
            equity=1_000_000,
            policy=policy(),
        )
        alpha, risk, cost, securities = inputs(first_cost_bps=250.0)
        high_cost = build_baseline_pipeline(
            alpha=alpha,
            risk=risk,
            cost=cost,
            securities=securities,
            equity=1_000_000,
            policy=policy(),
        )
        symbol = SYMBOLS[0]
        self.assertLess(
            high_risk.solution.weights[symbol],
            base.solution.weights[symbol],
        )
        self.assertLess(
            high_cost.solution.weights[symbol],
            base.solution.weights[symbol],
        )
        self.assertNotEqual(
            base.layer_trace[2]["event_sha256"],
            high_risk.layer_trace[2]["event_sha256"],
        )
        self.assertNotEqual(
            base.layer_trace[2]["event_sha256"],
            high_cost.layer_trace[2]["event_sha256"],
        )

    def test_missing_layer_input_fails_closed(self):
        alpha, risk, cost, securities = inputs()
        with self.assertRaisesRegex(
            BaselinePipelineError,
            "symbol sets must match exactly",
        ):
            build_baseline_pipeline(
                alpha=alpha,
                risk=risk[:-1],
                cost=cost,
                securities=securities,
                equity=1_000_000,
                policy=policy(),
            )

    def test_smoke_model_cannot_claim_baseline_candidate(self):
        alpha, risk, cost, securities = inputs()
        alpha = [
            AlphaForecastInput(
                item.symbol,
                item.expected_excess_return,
                "portable-momentum-v1",
                item.feature_version,
            )
            for item in alpha
        ]
        with self.assertRaisesRegex(
            BaselinePipelineError,
            "platform smoke strategy",
        ):
            build_baseline_pipeline(
                alpha=alpha,
                risk=risk,
                cost=cost,
                securities=securities,
                equity=1_000_000,
                policy=policy(),
            )

    def test_layer_trace_tampering_is_rejected(self):
        alpha, risk, cost, securities = inputs()
        result = build_baseline_pipeline(
            alpha=alpha,
            risk=risk,
            cost=cost,
            securities=securities,
            equity=1_000_000,
            policy=policy(),
        )
        trace = copy.deepcopy(list(result.layer_trace))
        trace[2]["output_sha256"] = "0" * 64
        with self.assertRaisesRegex(
            BaselinePipelineError,
            "output hash mismatch",
        ):
            verify_layer_trace(trace, result.layer_payloads)


if __name__ == "__main__":
    unittest.main()