feat(v2): T04 模拟案例生成与数据集划分
先边界 fixture 再批量生成；全程程序组合场景，**不做任何模型调用**（14 §8 末）。 新增 services/simulation_dataset.py： - generate(seed, scale)：程序化生成虚构项目，同 seed 同版本**逐条一致** - split()：按关联组 + 时间分桶（组内不拆），以组的最早发生时间排序 - assert_group_integrity / assert_source_allowed：组不跨集合；保留集来源禁止用于提炼 - manifest()：构造 DatasetManifest 契约体 契约补 T04 才该有的部分（T00/04 当初有意推迟）：EventOutcome 七值枚举、 OutcomeCount、DatasetManifest。含两条硬约束——合成数据必须 demo 域（BR-34）、 计数不得为 0（空结果用不适用表达）。 配比按 14 §8：通过 1800（直接 900 + 补件后 900）/ 未通过 600 / 不计入 600 （附条件 100 + 未送审 175 + 退出 175 + 待定 150）= 3000。 两个建模决定： - "补件后通过"是 approved 的**子类型**（outcome_subtype），不是独立 outcome —— 06 §3 要求区分首次提交与补件后，但设计 §9.1 的 EventOutcome 只有 7 个值，不自扩充 - conditional 是否计入目标由客户映射决定（OQ-05），故 label=None，绝不默认当通过 实现中被自己的检查抓到两个 bug： 1. 分组与近重复用了两套独立公式，导致 near_duplicate_of 指向别组记录， assert_group_integrity 直接报"近重复案例跨集合"。现组号与近重复由同一序号唯一推出 2. split() 末段取剩余可能为负（极端小样本），会静默切出错误的划分。已夹到非负 测试：新增 test_simulation_dataset.py 24 项（边界 fixture 10 + 生成 10 + 清单 4 + 落盘 2），含 AC-35 的可复现性与 AC-36 的组不跨集合/保留集来源拒绝。 全量 272 项 OK（248 → 272）；verify.py 10/10 PASS。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Code
@@ -0,0 +1,256 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user