"""T04 模拟案例生成与数据集划分。 依据:docs/v2/14-experience-loop.md §8(配比与隔离)、§4(划分先于经验提炼); 契约 owner:docs/v2/03-detailed-design.md §9.2 的 DatasetManifest(数据集服务)。 **边界先行**:先用手工 fixture 固化逻辑(0 例、单组、近重复、跨时间边界), 再按 14 §8 的配比批量生成。生成全程用程序组合结构化场景,**不做任何模型调用** (14 §8 末:不为每条案例进行多轮昂贵模型调用)。 隔离要求(BR-35 / AC-36): - 关联组(同客户/同集团/近重复)**整体**落入同一划分,不跨集合 - 时间划分按组的最早发生时间分桶,不按单条案例切 - 保留集及其衍生知识不得进入开发;越界即判定泄漏、该知识作废 不声称真实效果:所有记录标 synthetic/demo,概览文案不得写成准确率。 """ from __future__ import annotations import json import random from datetime import datetime, timedelta, timezone from pathlib import Path from services import assistant_store as store GENERATOR_VERSION = "simulation_dataset@1.0.0" TEMPLATE_VERSION = "xingbang-sim-templates@1.0.0" SPLIT_POLICY_VERSION = "group-time-split@1.0.0" SPLITS = ("development", "calibration", "holdout") # 时间划分比例(按组的代表时间分桶,14 §8:划分先于经验提炼) SPLIT_RATIOS = (("development", 0.6), ("calibration", 0.2), ("holdout", 0.2)) # 结果配比(docs/v2/14-experience-loop.md §8)。总计 3000 是**可调整的演示规模**, # 不是效果样本量承诺。 # # 两个刻意的建模决定: # 1. "补件后通过" 是 approved 的**子类型**,不是独立 outcome——06 §3 要求区分 # "首次提交通过"与"补件后最终通过",但事件编码本身仍是 approved(设计 §9.1 的 # EventOutcome 只有 7 个值,不得自行扩充)。 # 2. conditional(附条件通过)是否计入目标由客户映射决定(OQ-05),故 label=None, # 绝不默认当 approved(E-26 / BR-21)。 TEMPLATES = ( {"outcome": "approved", "subtype": "direct", "count": 900, "label": True, "desc": "材料齐全,评审会直接通过"}, {"outcome": "approved", "subtype": "after_supplement", "count": 900, "label": True, "desc": "补件后通过;初始材料不足"}, {"outcome": "rejected", "subtype": "direct", "count": 600, "label": False, "desc": "评审会未通过"}, {"outcome": "conditional", "subtype": "conditional", "count": 100, "label": None, "desc": "附条件通过,是否计入目标由映射决定"}, {"outcome": "not_submitted", "subtype": "not_submitted", "count": 175, "label": None, "desc": "未送审,不当作否决"}, {"outcome": "withdrawn", "subtype": "withdrawn", "count": 175, "label": None, "desc": "主动退出,不当作否决"}, {"outcome": "deferred", "subtype": "deferred", "count": 150, "label": None, "desc": "会议待定,口径未确认"}, ) FIELDS_SHAPE = { "facility.amount": ("600", "2400"), # 万元 "lease.total_value": ("500", "2000"), "financial.total_assets": ("3000", "40000"), "financial.total_liabilities": ("1000", "30000"), "financial.operating_cashflow": ("-800", "6000"), } def _text(value) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _counts(records: list[dict]) -> list[dict]: tally: dict[str, int] = {} for record in records: tally[record["outcome"]] = tally.get(record["outcome"], 0) + 1 return [{"outcome": key, "count": tally[key]} for key in sorted(tally)] def generate(seed: int = 20260929, scale: float = 1.0, templates=TEMPLATES) -> list[dict]: """按配比生成虚构项目。同 seed + 同 templates → 逐字节一致。 scale 只按比例缩放各档数量(便于小规模演示与测试),不改变配比结构。 """ if seed < 0: raise ValueError("seed 必须为非负整数") if scale <= 0: raise ValueError("scale 必须为正数") rng = random.Random(seed) records: list[dict] = [] base_time = datetime(2023, 1, 2, tzinfo=timezone.utc) serial = 0 for template in templates: wanted = max(1, int(round(template["count"] * scale))) for _ in range(wanted): serial += 1 # 关联组:每 3 条构成一组(同客户多项目 / 近重复),模拟"关联组不得跨集合"。 # 组号必须由记录序号**唯一**推出,且 near_duplicate_of 只能指向同组记录—— # 早先版本里分组与近重复用了两套独立公式,导致近重复落到别组(integrity 检查抓到)。 group_index = (serial - 1) // 3 + 1 near = f"SIM-{serial - 1:06d}" if serial % 3 == 2 else None occurred = base_time + timedelta(days=rng.randint(0, 900), hours=rng.randint(0, 23)) facts = { field: str(round(rng.uniform(float(low), float(high)), 2)) for field, (low, high) in FIELDS_SHAPE.items() } records.append({ "id": f"SIM-{serial:06d}", "synthetic": True, "data_domain": "demo", "customer": f"虚构客户{group_index:05d}", "customer_group": f"CG-{group_index:05d}", "near_duplicate_of": near, "occurred_at": occurred.isoformat(), "outcome": template["outcome"], "outcome_subtype": template["subtype"], # 未送审/待定/附条件一律 null:绝不默认当通过(BR-21 / E-26) "label": template["label"], "initial_facts": facts, "trace": { "generator_version": GENERATOR_VERSION, "template_version": TEMPLATE_VERSION, "seed": seed, "template": f'{template["outcome"]}/{template["subtype"]}', }, }) return records def source_hash(records: list[dict]) -> str: """对**规范化**后的记录取哈希,用于证明两次生成一致(AC-35)。""" normalized = [{k: v for k, v in sorted(record.items())} for record in records] normalized.sort(key=lambda item: item["id"]) return store.digest(normalized) def split(records: list[dict], policy_version: str = SPLIT_POLICY_VERSION) -> dict: """按关联组 + 时间分桶。 组的代表时间取该组**最早**的 occurred_at,然后按整体时间轴切三段。 以组为单位分配,因此天然不会把同客户/近重复拆到不同集合。 """ if not records: raise ValueError("没有可划分的案例") groups: dict[str, list[dict]] = {} for record in records: groups.setdefault(record["customer_group"], []).append(record) members = [] for group_id, items in groups.items(): representatives = sorted(item["occurred_at"] for item in items) # 近重复必须同组:分组键相同即为同组,这里再校验一次 members.append((group_id, representatives[0], representatives[-1], items)) members.sort(key=lambda item: (item[1], item[0])) total = len(members) boundaries, cursor = {}, 0 for index, (name, ratio) in enumerate(SPLIT_RATIOS): # 末段取剩余,避免四舍五入后总和超过组数;并夹到非负, # 否则极端小样本下会产生负切片而静默给出错误划分。 take = total - cursor if index == len(SPLIT_RATIOS) - 1 else int(round(total * ratio)) take = max(0, take) boundaries[name] = members[cursor:cursor + take] cursor += take assignments: dict[str, str] = {} for name, chunk in boundaries.items(): for group_id, _, _, items in chunk: assignments[group_id] = name for item in items: assignments[item["id"]] = name assert_group_integrity(records, assignments) return { "split_policy_version": policy_version, "assignments": assignments, "group_of": {item["id"]: item["customer_group"] for item in records}, "group_counts": {name: len(chunk) for name, chunk in boundaries.items()}, } def assert_group_integrity(records: list[dict], assignments: dict[str, str]) -> None: """按组成员逐条校验:同组所有案例必须落在同一集合。""" by_group: dict[str, set] = {} for record in records: by_group.setdefault(record["customer_group"], set()).add(assignments[record["id"]]) broken = {group: names for group, names in by_group.items() if len(names) > 1} if broken: raise ValueError(f"存在跨集合的关联组:{sorted(broken)[:5]}") for record in records: duplicate = record.get("near_duplicate_of") if duplicate and assignments.get(duplicate) != assignments[record["id"]]: raise ValueError(f"近重复案例跨集合:{record['id']} 与 {duplicate}") def assert_source_allowed(split_result: dict, source_ids, target_split: str) -> None: """衍生知识不得使用保留集来源(BR-35 / AC-36)。 越界即判定泄漏:调用方必须作废该候选,而不是降级使用。 """ assignments = split_result["assignments"] if target_split == "holdout": raise ValueError("保留集自身不参与经验提炼") offenders = [source_id for source_id in source_ids if assignments.get(source_id) == "holdout"] if offenders: raise ValueError(f"来源属于保留集,禁止用于提炼并须作废:{sorted(offenders)[:5]}") def manifest(records: list[dict], split_result: dict, seed: int | None = None, split_id: str = "p0-demo") -> dict: """构造 DatasetManifest 契约体(03 §9.2)。""" from contracts.prescreen import DatasetManifest payload = { "id": store.uid("dataset"), "version": f"{GENERATOR_VERSION}+{TEMPLATE_VERSION}", "data_domain": "demo", "seed": seed, "generator_version": GENERATOR_VERSION, "template_version": TEMPLATE_VERSION, "source_hash": source_hash(records), "split_id": split_id, "split_policy_version": split_result["split_policy_version"], "group_assignments_ref": store.digest(split_result["assignments"]), "counts": _counts(records), "created_at": store.now(), "synthetic": True, } return DatasetManifest(**payload).model_dump(mode="json") def assert_labelled_outcomes_are_mapped(records: list[dict]) -> None: """有 label 的结果只能是明确映射过的(BR-21)。""" allowed = {"approved", "approved_after_supplement", "rejected"} for record in records: if record["label"] is not None and record["outcome"] not in allowed: raise ValueError(f"结果 {record['outcome']} 尚未确认口径,不得带标签") def write(records: list[dict], directory: Path, name: str = "simulation.jsonl") -> Path: """落盘到指定目录。**默认不入 Git**:调用方应指向被 gitignore 的数据根。""" if "xingbang_data" not in str(directory): raise ValueError("模拟数据只写入 xingbang_data 数据根(该路径已被 gitignore)") directory.mkdir(parents=True, exist_ok=True) target = directory / name with target.open("w", encoding="utf-8", newline="\n") as handle: for record in records: handle.write(_text(record) + "\n") return target