"""多 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"]))