tests/test_condition_selection.py

"""目标储存条件选择的回归测试（用户反馈：长期预测被误用加速数据外推）。

场景还原（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"]))