"""目标储存条件选择的回归测试（用户反馈：长期预测被误用加速数据外推）。 场景还原（Demo2 data.docx）：每批含 25℃/60%RH 长期（多点至 36 月）与 40℃/75%RH 加速（仅 3 点至 6 月）两条件。用户指令"预测 25℃,60%RH 条件下 48 个月降解产物"。 修复前缺陷： - 目标时间点解析把"25℃""60%RH"中的 25/60 误当成月份； - 决策引擎对每个条件都生成预测并按时间点 key 覆盖，最终保留**最后处理的加速条件**， 导致用加速 6 月数据外推 48 月（is_valid=False、点估计虚高）。 修复后应： - 仅解析出目标时间点 [48]； - 捕获用户目标条件（25℃/长期）； - 预测取**长期 25℃** 拟合，48 月落在 ICH 2×（36 月窗）以内 → 有效。 导入路径由 tests/conftest.py 设置（底座以顶层包导入；skills 为顶层包）。 """ import pytest from skills.stability.skill import ( _target_timepoints_from_goal, _target_condition_from_goal, _build_intent_dict, _build_intent_object, _primary_series, ) GOAL = "请梳理附件稳定性数据并预测各批样品25℃，60%RH条件下48个月时的降解产物" def _demo_data(): """Demo2 风格数据：长期 25℃（8 点至 36 月）+ 加速 40℃（3 点至 6 月）。""" return { "batches": [{ "batch_id": "Batch_1", "batch_name": "Batch_1", "batch_type": "target", "conditions": [ {"condition_id": "25C_60RH", "condition_type": "longterm", "timepoints": [0, 3, 6, 9, 12, 18, 24, 36], "cqa_data": [{"cqa_name": "降解产物", "spec_type": "upper", "values": [0.1, 0.1, 0.1, 0.2, 0.2, 0.3, 0.4, 0.6]}]}, {"condition_id": "40C_75RH", "condition_type": "accelerated", "timepoints": [0, 3, 6], "cqa_data": [{"cqa_name": "降解产物", "spec_type": "upper", "values": [0.1, 0.4, 0.8]}]}, ], }], "specification_limit": 0.5, "primary_cqa": "降解产物", "target_timepoints": [48], } def test_timepoints_not_polluted_by_temp_humidity(): """'25℃，60%RH...48个月' 只应解析出 [48]，不得混入 25 / 60。""" tps = _target_timepoints_from_goal(GOAL) assert tps == [48] assert 25 not in tps and 60 not in tps def test_target_condition_parsed_as_longterm_25c(): cond = _target_condition_from_goal(GOAL) assert cond is not None assert cond["temp_c"] == pytest.approx(25.0) assert cond["rh"] == pytest.approx(60.0) assert cond["condition_type"] == "longterm" def test_intent_dict_carries_target_condition(): intent = _build_intent_dict(GOAL, _demo_data()) assert intent["target_timepoints"] == [48] assert intent["target_condition"]["condition_type"] == "longterm" def test_engine_predicts_from_longterm_not_accelerated(): """决策引擎应使用 25℃ 长期数据预测，48 月落在 ICH 2× 内 → 有效。""" from layers.regulatory_decision_engine import RegulatoryDecisionEngine data = _demo_data() intent_dict = _build_intent_dict(GOAL, data) data["target_condition"] = intent_dict["target_condition"] intent = _build_intent_object(intent_dict, data) result = RegulatoryDecisionEngine().execute(intent, data) # 两条件都应被拟合（用于展示），但预测只来自长期条件。 assert "25C_60RH" in result.kinetic_fits assert "40C_75RH" in result.kinetic_fits preds = result.predictions assert "48M" in preds p = preds["48M"] # 长期 k≈0.0146 → 48 月点估计远低于加速误用时的 ~5.68%。 assert p.point_estimate < 1.5, f"应来自长期数据的温和外推，实际 {p.point_estimate}" # 48 月在长期 36 月窗的 2× 内 → 有效外推。 assert p.is_valid is True # 审计追踪记录所选预测条件。 steps = {e["step"]: e for e in result.calculation_trace.entries} assert "select_prediction_condition" in steps assert steps["select_prediction_condition"]["outputs"]["selected_condition"] == "25C_60RH" def test_primary_series_uses_longterm_condition(): """建模 / 绘带的主序列也应取长期 25℃ 条件（与预测口径一致）。""" data = _demo_data() intent_dict = _build_intent_dict(GOAL, data) data["target_condition"] = intent_dict["target_condition"] series = _primary_series(data) assert series is not None # 长期条件有 8 个点；加速仅 3 点。取到长期则点数为 8。 assert len(series["times"]) == 8 assert max(series["times"]) == 36 def test_no_target_condition_defaults_to_longterm(): """未显式指定条件时，仍应默认用长期条件预测（科学正确的货架期口径）。""" from layers.regulatory_decision_engine import RegulatoryDecisionEngine data = _demo_data() # 目标不含条件描述。 goal = "预测各批样品48个月的降解产物" intent_dict = _build_intent_dict(goal, data) assert intent_dict.get("target_condition") is None intent = _build_intent_object(intent_dict, data) result = RegulatoryDecisionEngine().execute(intent, data) steps = {e["step"]: e for e in result.calculation_trace.entries} assert steps["select_prediction_condition"]["outputs"]["selected_condition"] == "25C_60RH" # --------------------------------------------------------------------------- # 逐批次预测（用户反馈：报告应给出每一批的预测结果） # --------------------------------------------------------------------------- def _multi_batch_demo_data(): """3 批次，各含长期 25℃（8 点至 36 月），降解速率递增以体现批间差异。""" def batch(bid, vals): return { "batch_id": bid, "batch_name": bid, "batch_type": "target" if bid == "Batch_1" else "reference", "conditions": [{ "condition_id": "25C_60RH", "condition_type": "longterm", "timepoints": [0, 3, 6, 9, 12, 18, 24, 36], "cqa_data": [{"cqa_name": "降解产物", "spec_type": "upper", "values": vals}], }], } return { "batches": [ batch("Batch_1", [0.1, 0.1, 0.1, 0.2, 0.2, 0.3, 0.4, 0.6]), batch("Batch_2", [0.1, 0.1, 0.2, 0.2, 0.3, 0.4, 0.5, 0.7]), batch("Batch_3", [0.1, 0.2, 0.2, 0.3, 0.3, 0.4, 0.6, 0.8]), ], "specification_limit": 0.5, "primary_cqa": "降解产物", "target_timepoints": [48], } def test_per_batch_predictions_cover_all_batches(): """compute 应为每个批次给出目标时间点预测（呈现批间差异）。""" from kernel.skill_base import ExtractedData from skills.stability.skill import StabilitySkill data = _multi_batch_demo_data() data["target_condition"] = {"temp_c": 25.0, "rh": 60.0, "condition_type": "longterm"} payload = { "data": data, "intent": _build_intent_dict("预测各批长期48个月降解产物", data), "extraction_method": "manual", } cr = StabilitySkill().compute(ExtractedData(payload=payload, method="manual")) assert cr.can_proceed pbp = cr.summary.get("per_batch_predictions") assert pbp is not None, "应产出逐批次预测" batches = {r["batch"] for r in pbp["rows"]} assert batches == {"Batch_1", "Batch_2", "Batch_3"}, "三个批次都应有预测" # 目标时间点对齐意图（48 月）。 assert pbp["target_timepoints"] == [48] # 批间差异：降解更快的批次点估计更高。 by_batch = {r["batch"]: r["point_estimate"] for r in pbp["rows"] if r["timepoint"] == 48} assert by_batch["Batch_1"] < by_batch["Batch_2"] < by_batch["Batch_3"] def test_per_batch_predictions_rendered_in_report(): """report.assemble 应渲染逐批次预测表。""" from services.report_service import ReportService from kernel.skill_base import ReportSections class _Meta: id = "stability"; display_name = "稳定性预测"; version = "1.0.0" summary = { "per_batch_predictions": { "cqa": "降解产物", "condition": "25C_60RH", "spec_provided": False, "spec_limit": None, "target_timepoints": [48], "rows": [ {"batch": "Batch_1", "condition": "25C_60RH", "timepoint": 48, "point_estimate": 0.7531, "CI_lower": 0.6394, "CI_upper": 0.8669, "is_valid": True, "R2": 0.9744, "model": "zero-order"}, {"batch": "Batch_2", "condition": "25C_60RH", "timepoint": 48, "point_estimate": 0.9132, "CI_lower": 0.8275, "CI_upper": 0.9989, "is_valid": True, "R2": 0.9897, "model": "zero-order"}, ], }, } class _Result: def __init__(self): self.summary = summary; self.figures = {} html = ReportService().assemble(_Meta(), _Result(), ReportSections(sections={}), lang="zh") assert "逐批次预测" in html assert "Batch_1" in html and "Batch_2" in html assert "0.7531" in html and "0.9132" in html if __name__ == "__main__": # pragma: no cover import sys sys.exit(pytest.main([__file__, "-v"]))