218 lines
10 KiB
Python
218 lines
10 KiB
Python
"""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 _target(key, dimension_ids):
|
||
aliases = {'assume_bank_flow': ('DC-CASH', 'corporate.cashflow'),
|
||
'assume_title': ('DC-TITLE', 'corporate.title')}
|
||
return next((name for name in aliases[key] if name in set(dimension_ids)), None)
|
||
|
||
|
||
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 = dict(ASSUMPTIONS[key], dimension_id=_target(key, known) or ASSUMPTIONS[key]["dimension_id"])
|
||
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": _target(key, known),
|
||
"assumed_stage": item["assumed_stage"], "premise": item["premise"]}
|
||
for key, item in ASSUMPTIONS.items() if _target(key, 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("情景基线副本的真实估计必须为空")
|