"""T09 推进沙盘服务：把"假设补件"算成独立情景，不触碰真实运行。 依据：docs/v2/13-probability-prototype.md §2/§4、04-ui-spec.md §5/§9、 03-detailed-design.md §9.3；BR-11/29/30、E-21/22、AC-11/12/30/31/44。 设计要点： 1. **演示值查表，不累加**。结果由完整动作组合选出（`selection_hash`）。表里 00/10/01/11 是四条独立条目，不是 42+16+9=67 的算术和；`assert_not_additive` 把这条约束写成可执行断言，防止后来者把表换成"每项加分"。 2. **真实运行只读**。`simulate()` 接受运行视图 dict，返回新对象；基线的演示值 以副本随行携带，任何时候都不写回运行的 `probability`。 3. **越域即拒绝**。production 域不得有演示估计（E-21）；假设指向不存在的维度、 同一维度两个互斥假设，都按 E-22 拒绝，而不是忽略掉一个。 """ from __future__ import annotations import hashlib import json from contracts.prescreen import DataDomain, ErrorCode, ProbabilityView from contracts.simulation import Simulation, SimulationAssumption from services import assistant_store as store DEMO_TABLE_VERSION = "DEMO-TABLE-v1" DEMO_SOURCE = "synthetic_demo" # 说明文字固定：界面必须紧邻数值展示，避免被当作真实概率（DESIGN.md §38） DEMO_LABEL = "人工设定，仅演示交互；未经历史数据校准" STAGE_NAMES = ("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核") class DemoEstimateForbidden(ValueError): """当前数据域不允许使用演示概率（E-21）。""" class AssumptionConflict(ValueError): """所选推进条件冲突、缺少前提或指向不存在的维度（E-22）。""" # 假设动作登记表。`dimension_id` 必须与冻结方案里的维度 id 一致，否则拒绝。 ASSUMPTIONS: dict[str, dict] = { "assume_bank_flow": { "label": "假设银行流水逐笔匹配", "dimension_id": "DC-CASH", "assumed_stage": 3, "premise": "假设核验支持账期变化解释；只改变回款核验维度", }, "assume_title": { "label": "假设设备权属证明一致", "dimension_id": "DC-TITLE", "assumed_stage": 3, "premise": "假设合同、发票与设备清单核验一致；只改变权属维度", }, } # 演示情景表（13 §3）。key = 流水假设 + 权属假设的组合位。 DEMO_SCENARIOS: dict[str, dict] = { "00": {"value": 42, "remaining": ["回款核验仍待核实", "设备权属仍待核实"], "reason": "未选择假设。回款与设备权属证据仍有缺口，保持演示基线。"}, "10": {"value": 58, "remaining": ["设备权属仍待核实"], "reason": "假设银行流水逐笔核验后支持账期变化解释，回款不确定性减小；" "设备权属缺口仍在。演示变化 +16 个百分点。"}, "01": {"value": 51, "remaining": ["实际回款仍待核实"], "reason": "假设合同、发票与清单一致且经核验，权属争议得到回应；" "实际回款仍缺银行流水。演示变化 +9 个百分点。"}, "11": {"value": 64, "remaining": ["仍需整体人工复核"], "reason": "假设回款与权属核验均支持项目解释。组合情景为 64%，变化 +22 " "个百分点；组合影响不简单相加，仍需综合复核其他风险。"}, } def scenario_key(keys) -> str: """把动作组合压成表键：按登记顺序取位，未选为 0。""" order = list(ASSUMPTIONS) unknown = sorted(set(keys or ()) - set(order)) if unknown: raise AssumptionConflict(f"未登记的假设动作：{unknown}") return "".join("1" if item in set(keys or ()) else "0" for item in order) def selection_hash(run_id: str, revision: int, keys) -> str: """完整动作组合的摘要。前端只接受与当前选择一致的结果（AC-31）。""" payload = json.dumps({"run_id": run_id, "revision": revision, "key": scenario_key(keys)}, ensure_ascii=False, sort_keys=True) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def assert_demo_domain(data_domain) -> None: value = data_domain.value if hasattr(data_domain, "value") else str(data_domain) if value != DataDomain.demo.value: raise DemoEstimateForbidden("当前数据域不允许使用演示概率；真实概率需完成校准后另立验证") def assert_not_additive() -> None: """组合值必须来自查表：任何"单项增量相加"的实现都会在这里失败。""" baseline = DEMO_SCENARIOS["00"]["value"] pair = DEMO_SCENARIOS["11"]["value"] added = baseline + (DEMO_SCENARIOS["10"]["value"] - baseline) \ + (DEMO_SCENARIOS["01"]["value"] - baseline) if pair == added: raise AssertionError("演示情景表退化成单项加和，须回到按完整组合查表（BR-30）") def _view(snapshot_id: str, key: str, as_of: str | None = None) -> ProbabilityView: row = DEMO_SCENARIOS[key] return ProbabilityView( snapshot_id=snapshot_id, as_of=as_of or store.now(), demo_estimate={"value": row["value"] / 100, "source": DEMO_SOURCE, "scenario_table_version": DEMO_TABLE_VERSION, "scenario_key": key, "label": DEMO_LABEL}) def _assumptions_for(keys, dimension_ids) -> list[SimulationAssumption]: known = set(dimension_ids) chosen: list[SimulationAssumption] = [] for key in ASSUMPTIONS: if key not in set(keys or ()): continue item = ASSUMPTIONS[key] if item["dimension_id"] not in known: raise AssumptionConflict( f"假设动作 {key} 指向本次方案不存在的维度 {item['dimension_id']}") chosen.append(SimulationAssumption( key=key, label=item["label"], target_dimension_id=item["dimension_id"], assumed_stage=item["assumed_stage"], premise=item["premise"])) return chosen def _dimension_states(run_view: dict, assumptions) -> list[dict]: """只覆盖被假设的维度；其余维度原样保留并标为未变（13 §P-17）。""" assumed = {item.target_dimension_id: item for item in assumptions} rows = [] for item in run_view.get("dimension_results") or (): hit = assumed.get(item["id"]) rows.append({"id": item["id"], "baseline_stage": item.get("stage"), "scenario_stage": hit.assumed_stage if hit else item.get("stage"), "assumed": hit is not None}) return rows def failed(run_view: dict, keys, message: str, code: ErrorCode, revision: int | None = None) -> dict: """情景失败：保留基线、标未完成，不得回退成"演示表默认值"（13 §44）。""" revision = revision or int(run_view.get("revision") or 1) baseline = _baseline_view(run_view) body = Simulation( id=store.uid("simulation"), run_id=run_view["id"], project_id=run_view["project_id"], data_domain=run_view["data_domain"], revision=revision, selection_hash=selection_hash(run_view["id"], revision, keys), status="failed", assumptions=[], baseline_probability_view=baseline, probability_view=None, dimensions=_dimension_states(run_view, []), remaining_inputs=list(run_view.get("missing_inputs") or ()), reason=message, error_code=code, frozen_versions=dict(run_view.get("versions") or ()), created_at=store.now()) return body.model_dump(mode="json") def available_assumptions(run_view: dict) -> list[dict]: """列出本次方案可用的假设动作。 只返回目标维度真实存在于该运行的动作：界面因此不会给出点了必然失败的选项， 服务端也不靠"忽略无效动作"来掩饰不匹配（E-22）。 """ known = {item["id"] for item in run_view.get("dimension_results") or ()} return [{"key": key, "label": item["label"], "dimension_id": item["dimension_id"], "assumed_stage": item["assumed_stage"], "premise": item["premise"]} for key, item in ASSUMPTIONS.items() if item["dimension_id"] in known] def _baseline_view(run_view: dict) -> ProbabilityView: snapshot_id = run_view.get("snapshot_id") or run_view.get("input_snapshot_id") if not snapshot_id: raise ValueError("运行缺少输入快照，无法建立情景基线") return _view(snapshot_id, "00") def simulate(run_view: dict, keys, revision: int | None = None) -> dict: """按完整动作组合产出一条情景；真实运行只被读取，不被改写。""" assert_demo_domain(run_view["data_domain"]) revision = revision or int(run_view.get("revision") or 1) dimension_ids = [item["id"] for item in run_view.get("dimension_results") or ()] assumptions = _assumptions_for(keys, dimension_ids) key = scenario_key(keys) row = DEMO_SCENARIOS[key] baseline = _baseline_view(run_view) body = Simulation( id=store.uid("simulation"), run_id=run_view["id"], project_id=run_view["project_id"], data_domain=run_view["data_domain"], revision=revision, selection_hash=selection_hash(run_view["id"], revision, keys), status="ready", assumptions=assumptions, baseline_probability_view=baseline, probability_view=_view(baseline.snapshot_id, key, baseline.as_of), dimensions=_dimension_states(run_view, assumptions), remaining_inputs=list(row["remaining"]), reason=row["reason"], frozen_versions=dict(run_view.get("versions") or ()), created_at=store.now()) return body.model_dump(mode="json") def assert_baseline_untouched(run_view: dict, simulation: dict) -> None: """装配出口的自检：情景不得改写运行的概率字段（BR-11 / AC-11）。""" if run_view.get("probability") is not None: raise AssertionError("情景推演不得给真实运行写入概率") if (run_view.get("probability_view") or {}).get("real_estimate") is not None: raise AssertionError("情景推演不得给真实运行写入真实估计") baseline = simulation.get("baseline_probability_view") or {} if baseline.get("real_estimate") is not None: raise AssertionError("情景基线副本的真实估计必须为空")