tests/test_multi_cqa.py

"""多 CQA 木桶原理联合决策单元测试（任务 15 / 需求 11）。

覆盖需求：
- 11.1：同时支持上限型规格（杂质，越界为超上限）与下限型规格（含量，越界为低于下限）。
- 11.2：分别对每个 CQA 计算其满足规格的最长时间。
- 11.3：最终货架期 = min(各 CQA 单独货架期)（木桶原理），并标明决定性 CQA。
- 11.4：下限型 CQA 风险判定基于预测区间**下界**与下限的关系（而非上界与上限）。
- 11.5：各 CQA 单独结论与联合结论可区分呈现（per_cqa + 联合结果）。

纯 Python 计算，不调用 LLM（设计 Property 1）。导入路径由 ``tests/conftest.py``
与仓库根 ``conftest.py`` 统一设置。
"""

from __future__ import annotations

import sys

import numpy as np
import pytest

from skills.stability.stability_calculator import StabilityCalculator
from skills.stability.multi_cqa import (
    cqa_shelf_life_from_fit,
    evaluate_multi_cqa,
    joint_shelf_life,
    _first_crossing,
    _normalize_spec_type,
)
from schemas.decision_result import (
    CQAShelfLifeResult,
    JointShelfLifeResult,
    SpecType,
)


@pytest.fixture
def calc() -> StabilityCalculator:
    return StabilityCalculator()


# 上升型杂质（上限型）：从 0.10% 缓慢上升，含轻微散度。
IMPURITY_TIMES = [0.0, 3.0, 6.0, 9.0, 12.0]
IMPURITY_VALUES = [0.10, 0.155, 0.205, 0.262, 0.305]  # ~0.10 + 0.017·t

# 快速上升型杂质（上限型）：更陡，应更早越上限 → 更短货架期。
FAST_IMPURITY_VALUES = [0.10, 0.21, 0.305, 0.41, 0.50]  # ~0.10 + 0.033·t

# 下降型含量（下限型）：从 100% 缓慢下降。
ASSAY_TIMES = [0.0, 3.0, 6.0, 9.0, 12.0]
ASSAY_VALUES = [100.0, 99.2, 98.5, 97.6, 96.9]  # ~100 − 0.26·t


# ---------------------------------------------------------------------------
# 需求 11.1 / 11.4：spec_type 归一化与下限型方向
# ---------------------------------------------------------------------------

def test_normalize_spec_type_variants():
    assert _normalize_spec_type("upper") == "upper"
    assert _normalize_spec_type("lower") == "lower"
    assert _normalize_spec_type("上限") == "upper"
    assert _normalize_spec_type("下限型") == "lower"
    assert _normalize_spec_type(SpecType.LOWER) == "lower"
    assert _normalize_spec_type(SpecType.UPPER) == "upper"
    # 缺省 / 未知 → 上限型（向后兼容既有杂质场景）。
    assert _normalize_spec_type("") == "upper"
    assert _normalize_spec_type(None) == "upper"


def test_first_crossing_upper_and_lower():
    times = [0.0, 1.0, 2.0, 3.0, 4.0]
    # 上限型：bound 上升，limit=2.5 → 在 t=2..3 之间穿越。
    up_bound = [0.0, 1.0, 2.0, 3.0, 4.0]
    tc, crossed = _first_crossing(times, up_bound, 2.5, "upper")
    assert crossed is True
    assert tc == pytest.approx(2.5, abs=1e-9)

    # 下限型：bound 下降，limit=1.5 → 在 t=2..3 之间穿越。
    low_bound = [4.0, 3.0, 2.0, 1.0, 0.0]
    tc2, crossed2 = _first_crossing(times, low_bound, 1.5, "lower")
    assert crossed2 is True
    assert tc2 == pytest.approx(2.5, abs=1e-9)

    # 从不越界 → 右删失，返回 horizon 末端。
    tc3, crossed3 = _first_crossing(times, [0.0, 0.1, 0.2, 0.3, 0.4], 10.0, "upper")
    assert crossed3 is False
    assert tc3 == pytest.approx(4.0)


# ---------------------------------------------------------------------------
# 需求 11.2 / 11.4：单 CQA 货架期由正确的预测区间边界决定
# ---------------------------------------------------------------------------

def test_upper_cqa_shelf_life_uses_ci_upper_bound(calc):
    """上限型 CQA：货架期由 CI 上界达到上限的时间决定，且早于点估计越界时间。"""
    fit = calc.fit_zero_order(IMPURITY_TIMES, IMPURITY_VALUES)
    res = cqa_shelf_life_from_fit(
        fit, cqa_name="总杂质", spec_type="upper",
        specification_limit=0.5, horizon_months=48.0,
    )
    assert res.spec_type == "upper"

    # 手动用连续带复算 CI 上界穿越时间。
    band = calc.compute_ci_band(fit, t_start=0.0, t_end=48.0, num=361, clip_negative=True)
    tc_upper, _ = _first_crossing(band["times"], band["upper"], 0.5, "upper")
    tc_point, _ = _first_crossing(band["times"], band["point"], 0.5, "upper")

    assert res.shelf_life_months == pytest.approx(tc_upper, abs=0.2)
    # 保守性：CI 上界越界早于点估计越界。
    assert res.shelf_life_months <= tc_point + 1e-6


def test_lower_cqa_shelf_life_uses_ci_lower_bound(calc):
    """下限型 CQA（含量）：货架期由 CI 下界达到下限的时间决定（需求 11.4）。"""
    fit = calc.fit_zero_order(ASSAY_TIMES, ASSAY_VALUES)
    res = cqa_shelf_life_from_fit(
        fit, cqa_name="含量", spec_type="lower",
        specification_limit=95.0, horizon_months=48.0,
    )
    assert res.spec_type == "lower"

    band = calc.compute_ci_band(fit, t_start=0.0, t_end=48.0, num=361, clip_negative=False)
    tc_lower, _ = _first_crossing(band["times"], band["lower"], 95.0, "lower")
    tc_point, _ = _first_crossing(band["times"], band["point"], 95.0, "lower")

    # 货架期应由**下界**决定，而非上界/点估计。
    assert res.shelf_life_months == pytest.approx(tc_lower, abs=0.2)
    # 保守性：CI 下界越界早于点估计越界。
    assert res.shelf_life_months <= tc_point + 1e-6


def test_cqa_meeting_spec_within_horizon_is_censored(calc):
    """考察窗内始终满足规格 → 标记右删失，货架期取封顶值。"""
    fit = calc.fit_zero_order(IMPURITY_TIMES, IMPURITY_VALUES)
    res = cqa_shelf_life_from_fit(
        fit, cqa_name="总杂质", spec_type="upper",
        specification_limit=5.0,  # 远高于任何预测 → 永不越界
        horizon_months=24.0,
    )
    assert res.censored is True
    assert res.shelf_life_months == pytest.approx(24.0)
    assert res.risk_level == "compliant"


# ---------------------------------------------------------------------------
# 需求 11.3：联合货架期 = min(各 CQA)，决定性 CQA 标注正确
# ---------------------------------------------------------------------------

def test_joint_shelf_life_is_min_and_labels_determining_cqa():
    """联合货架期取最短者，determining_cqa 指向最短板。"""
    a = CQAShelfLifeResult(cqa_name="含量", spec_type="lower",
                           specification_limit=95.0, shelf_life_months=30.0)
    b = CQAShelfLifeResult(cqa_name="总杂质", spec_type="upper",
                           specification_limit=0.5, shelf_life_months=18.0)
    c = CQAShelfLifeResult(cqa_name="水分", spec_type="upper",
                           specification_limit=3.0, shelf_life_months=42.0)

    joint = joint_shelf_life([a, b, c])
    assert isinstance(joint, JointShelfLifeResult)
    assert joint.joint_shelf_life_months == pytest.approx(18.0)
    assert joint.determining_cqa == "总杂质"
    assert len(joint.per_cqa) == 3


def test_evaluate_multi_cqa_min_across_cqa(calc):
    """端到端：多 CQA（含上限+下限）联合货架期 = min(各单独货架期)。"""
    cqa_inputs = [
        {
            "cqa_name": "总杂质", "times": IMPURITY_TIMES, "values": IMPURITY_VALUES,
            "spec_type": "upper", "specification_limit": 0.5,
        },
        {
            "cqa_name": "降解产物", "times": IMPURITY_TIMES, "values": FAST_IMPURITY_VALUES,
            "spec_type": "upper", "specification_limit": 0.5,
        },
        {
            "cqa_name": "含量", "times": ASSAY_TIMES, "values": ASSAY_VALUES,
            "spec_type": "lower", "specification_limit": 95.0,
        },
    ]
    joint = evaluate_multi_cqa(cqa_inputs, horizon_months=48.0)

    per = {c.cqa_name: c.shelf_life_months for c in joint.per_cqa}
    expected_min = min(per.values())
    expected_cqa = min(per, key=per.get)

    # 木桶原理：联合 = 各 CQA 最短者。
    assert joint.joint_shelf_life_months == pytest.approx(expected_min)
    assert joint.determining_cqa == expected_cqa

    # 更陡的降解产物应早于较缓的总杂质越上限。
    assert per["降解产物"] < per["总杂质"]


def test_evaluate_multi_cqa_lower_limit_can_be_determining(calc):
    """下限型 CQA 在下降足够快时可成为决定性短板。"""
    fast_assay = [100.0, 96.5, 93.0, 89.0, 85.0]  # 快速下降，~ -1.25/月
    slow_impurity = [0.10, 0.12, 0.14, 0.165, 0.185]  # 缓慢上升
    cqa_inputs = [
        {
            "cqa_name": "总杂质", "times": IMPURITY_TIMES, "values": slow_impurity,
            "spec_type": "upper", "specification_limit": 0.5,
        },
        {
            "cqa_name": "含量", "times": ASSAY_TIMES, "values": fast_assay,
            "spec_type": "lower", "specification_limit": 95.0,
        },
    ]
    joint = evaluate_multi_cqa(cqa_inputs, horizon_months=48.0)
    assert joint.determining_cqa == "含量"
    per = {c.cqa_name: c.shelf_life_months for c in joint.per_cqa}
    assert joint.joint_shelf_life_months == pytest.approx(min(per.values()))


def test_joint_requires_at_least_one_cqa():
    with pytest.raises(ValueError):
        joint_shelf_life([])


# ---------------------------------------------------------------------------
# 决策引擎集成：execute() 产出 joint_shelf_life（向后兼容既有消费者）
# ---------------------------------------------------------------------------

def _build_intent(primary_cqa: str, spec_limit: float):
    from schemas.analysis_intent import (
        AnalysisIntent, AnalysisType, UserPreferences, HardConstraints,
        ExtractedDataSummary,
    )
    return AnalysisIntent(
        raw_goal="预测多 CQA 货架期",
        analysis_type=AnalysisType.SHELF_LIFE_PREDICTION,
        preferences=UserPreferences(target_timepoints=[24], required_confidence=0.95),
        constraints=HardConstraints(primary_cqa=primary_cqa, specification_limit=spec_limit),
        data_summary=ExtractedDataSummary(),
    )


def _build_extracted_data():
    return {
        "batches": [
            {
                "batch_id": "B001",
                "batch_type": "target",
                "conditions": [
                    {
                        "condition_id": "25C_60RH",
                        "timepoints": IMPURITY_TIMES,
                        "cqa_data": [
                            {
                                "cqa_name": "总杂质",
                                "values": IMPURITY_VALUES,
                                "spec_type": "upper",
                                "specification_limit": 0.5,
                            },
                            {
                                "cqa_name": "降解产物",
                                "values": FAST_IMPURITY_VALUES,
                                "spec_type": "upper",
                                "specification_limit": 0.5,
                            },
                            {
                                "cqa_name": "含量",
                                "values": ASSAY_VALUES,
                                "spec_type": "lower",
                                "specification_limit": 95.0,
                            },
                        ],
                    }
                ],
            }
        ]
    }


def test_decision_engine_populates_joint_shelf_life():
    """决策引擎在多 CQA 数据下产出联合货架期，且 = min(各 CQA)。"""
    from layers.regulatory_decision_engine import RegulatoryDecisionEngine

    engine = RegulatoryDecisionEngine()
    intent = _build_intent(primary_cqa="总杂质", spec_limit=0.5)
    result = engine.execute(intent, _build_extracted_data())

    assert result.joint_shelf_life is not None
    joint = result.joint_shelf_life
    per = {c.cqa_name: c.shelf_life_months for c in joint.per_cqa}
    assert set(per) == {"总杂质", "降解产物", "含量"}
    assert joint.joint_shelf_life_months == pytest.approx(min(per.values()))
    assert joint.determining_cqa == min(per, key=per.get)

    # 计算追踪应记录联合决策步骤（审计）。
    steps = [e["step"] for e in result.calculation_trace.entries]
    assert "multi_cqa_joint_shelf_life" in steps


def test_decision_engine_backward_compatible_without_spec_type():
    """无 per-CQA spec 信息时，joint_shelf_life 为 None，既有单 CQA 流程不受影响。"""
    from layers.regulatory_decision_engine import RegulatoryDecisionEngine

    engine = RegulatoryDecisionEngine()
    intent = _build_intent(primary_cqa="总杂质", spec_limit=0.5)
    data = {
        "batches": [
            {
                "batch_id": "B001",
                "batch_type": "target",
                "conditions": [
                    {
                        "condition_id": "25C_60RH",
                        "timepoints": IMPURITY_TIMES,
                        "cqa_data": [
                            {"cqa_name": "总杂质", "values": IMPURITY_VALUES},
                        ],
                    }
                ],
            }
        ]
    }
    result = engine.execute(intent, data)
    assert result.joint_shelf_life is None
    # 既有单 CQA 拟合仍产生。
    assert result.can_proceed is True
    assert len(result.kinetic_fits) >= 1


if __name__ == "__main__":  # pragma: no cover
    sys.exit(pytest.main([__file__, "-v"]))